feat(warehouse-suggestions): add the suggestion model and review api

frontend/package.json

@@ -154,6 +154,7 @@
         "@posthog/products-tracing": "workspace:*",
         "@posthog/products-visual-review": "workspace:*",
         "@posthog/products-warehouse-sources": "workspace:*",
+        "@posthog/products-warehouse-suggestions": "workspace:*",
         "@posthog/products-web-analytics": "workspace:*",
         "@posthog/products-webmcp": "workspace:*",
         "@posthog/products-wizard": "workspace:*",

posthog/models/activity_logging/activity_log.py

@@ -130,6 +130,7 @@
     "TableCertification",
     "DataQualityCheck",
     "DataQualityCheckSchedule",
+    "WarehouseSuggestion",
     "Billing",
     "Loop",
     "StamphogRepoConfig",
@@ -733,6 +734,11 @@ def mask_change_values(scope: str, field: str, before: Any, after: Any) -> Maske
         "saved_query",
         "table",
     ],
+    "WarehouseSuggestion": [
+        "evidence",
+        "score_inputs",
+        "payload",
+    ],
     "Loop": [
         # FK relations are not JSON-serializable for the change detail (same reason
         # FeatureFlag/Subscription exclude theirs).

posthog/settings/web.py

@@ -126,6 +126,7 @@
     "products.data_quality.backend.apps.DataQualityConfig",
     "products.security.backend.apps.SecurityConfig",
     "products.webmcp.backend.apps.WebmcpConfig",
+    "products.warehouse_suggestions.backend.apps.WarehouseSuggestionsConfig",
 ]
 
 INSTALLED_APPS = [
@@ -688,6 +689,7 @@ def static_varies_origin(headers, path, url):
             "TraceNodeKindEnum": "products.ai_observability.backend.facade.contracts.TRACE_NODE_KINDS",
             "SignalSourceProduct": "products.signals.backend.enums.SIGNAL_SOURCE_PRODUCT_VALUES",
             "SignalSourceType": "products.signals.backend.enums.SIGNAL_SOURCE_TYPE_VALUES",
+            "DismissalReasonEnum": "products.signals.backend.views.SIGNAL_REPORT_DISMISSAL_REASON_CHOICES",
             "ErrorTrackingIssueSeverityRuleEnum": ["low", "medium", "high", "critical"],
             #
             # The choices come from a typing.Literal via get_args; there is no class.

posthog/test/repo_invariants/setup_receivers_baseline.txt

@@ -51,6 +51,7 @@ model_activity:products.tasks.backend.activity_logging.handle_loop_change
 model_activity:products.tracing.backend.activity_logging.handle_traces_retention_rule_activity
 model_activity:products.warehouse_sources.backend.activity_logging.handle_external_data_schema_change
 model_activity:products.warehouse_sources.backend.activity_logging.handle_external_data_source_change
+model_activity:products.warehouse_suggestions.backend.activity_logging.handle_warehouse_suggestion_activity
 post_delete:ee.vercel.receivers.delete_experiment_experimentation_item
 post_delete:ee.vercel.receivers.delete_resource_experimentation_item
 post_delete:posthog.caching.organization_serializer_cache._connect_invalidation.<locals>.receiver_fn

products/warehouse_suggestions/backend/activity_logging.py

@@ -0,0 +1,33 @@
+from typing import Any, Optional, cast
+
+from posthog.models import User
+from posthog.models.activity_logging.activity_log import AuditableScope, Detail, changes_between, log_activity
+from posthog.models.signals import model_activity_signal, mutable_receiver
+
+from .models import WarehouseSuggestion
+
+
+@mutable_receiver(model_activity_signal, sender=WarehouseSuggestion)
+def handle_warehouse_suggestion_activity(
+    sender: type,
+    scope: str,
+    before_update: Optional[WarehouseSuggestion],
+    after_update: WarehouseSuggestion,
+    activity: str,
+    user: Optional[User],
+    was_impersonated: bool = False,
+    **kwargs: Any,
+) -> None:
+    log_activity(
+        organization_id=None,
+        team_id=after_update.team_id,
+        user=user,
+        was_impersonated=was_impersonated,
+        item_id=str(after_update.id),
+        scope=scope,
+        activity=activity,
+        detail=Detail(
+            name=f"{after_update.kind} suggestion",
+            changes=changes_between(cast(AuditableScope, scope), previous=before_update, current=after_update),
+        ),
+    )

products/warehouse_suggestions/backend/apps.py

@@ -0,0 +1,10 @@
+from django.apps import AppConfig
+
+
+class WarehouseSuggestionsConfig(AppConfig):
+    name = "products.warehouse_suggestions.backend"
+    label = "warehouse_suggestions"
+    verbose_name = "Warehouse suggestions"
+
+    def ready(self) -> None:
+        from . import activity_logging  # noqa: F401, PLC0415

products/warehouse_suggestions/backend/facade/api.py

@@ -0,0 +1,144 @@
+"""Facade for warehouse_suggestions."""
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+from uuid import UUID
+
+from ..logic import suggestions
+from ..logic.access import SubjectAccess, visible_suggestions
+from ..logic.flags import is_warehouse_suggestions_enabled
+from ..models import WarehouseSuggestion
+from .contracts import (
+    SubjectEditAccessRequiredError,
+    Suggestion,
+    SuggestionDraft,
+    SuggestionNotFoundError,
+    SuggestionPage,
+    SuggestionReviewer,
+)
+from .enums import (
+    WarehouseSuggestionAssetOutcome,
+    WarehouseSuggestionDismissalReason,
+    WarehouseSuggestionKind,
+    WarehouseSuggestionStatus,
+    WarehouseSuggestionSubjectKind,
+)
+
+if TYPE_CHECKING:
+    from products.access_control.backend.facade.user_access_control import UserAccessControl
+
+__all__ = [
+    "dismiss_suggestion",
+    "get_suggestion",
+    "is_warehouse_suggestions_enabled",
+    "list_suggestions",
+    "resume_suggestion",
+    "upsert_suggestions",
+]
+
+
+def upsert_suggestions(team_id: int, drafts: Sequence[SuggestionDraft]) -> None:
+    suggestions.upsert_suggestions(team_id, drafts)
+
+
+def list_suggestions(
+    team_id: int,
+    user_access_control: "UserAccessControl",
+    *,
+    kind: WarehouseSuggestionKind | None,
+    status: WarehouseSuggestionStatus | None,
+    limit: int,
+    offset: int,
+) -> SuggestionPage:
+    visible, access = visible_suggestions(team_id, user_access_control, kind=kind, status=status)
+    page = visible[offset : offset + limit]
+    return SuggestionPage(count=visible.count(), results=[_to_contract(row, access) for row in page])
+
+
+def get_suggestion(team_id: int, user_access_control: "UserAccessControl", suggestion_id: UUID) -> Suggestion:
+    row, access = _visible_suggestion(team_id, user_access_control, suggestion_id)
+    return _to_contract(row, access)
+
+
+def dismiss_suggestion(
+    team_id: int,
+    user_access_control: "UserAccessControl",
+    suggestion_id: UUID,
+    *,
+    user_id: int,
+    reason: WarehouseSuggestionDismissalReason,
+    note: str | None,
+) -> Suggestion:
+    return _decide(
+        team_id,
+        user_access_control,
+        suggestion_id,
+        WarehouseSuggestionStatus.DISMISSED,
+        user_id=user_id,
+        reason=reason,
+        note=note,
+    )
+
+
+def resume_suggestion(
+    team_id: int, user_access_control: "UserAccessControl", suggestion_id: UUID, *, user_id: int
+) -> Suggestion:
+    return _decide(team_id, user_access_control, suggestion_id, WarehouseSuggestionStatus.PROPOSED, user_id=user_id)
+
+
+def _decide(
+    team_id: int,
+    user_access_control: "UserAccessControl",
+    suggestion_id: UUID,
+    new_status: WarehouseSuggestionStatus,
+    *,
+    user_id: int,
+    reason: WarehouseSuggestionDismissalReason | None = None,
+    note: str | None = None,
+) -> Suggestion:
+    row, access = _visible_suggestion(team_id, user_access_control, suggestion_id)
+    if not access.can_act_on(row):
+        raise SubjectEditAccessRequiredError(WarehouseSuggestionSubjectKind(row.subject_kind))
+    decided = suggestions.transition_to(row.id, team_id, new_status, user_id=user_id, reason=reason, note=note)
+    return _to_contract(decided, access)
+
+
+def _visible_suggestion(
+    team_id: int, user_access_control: "UserAccessControl", suggestion_id: UUID
+) -> tuple[WarehouseSuggestion, SubjectAccess]:
+    visible, access = visible_suggestions(team_id, user_access_control, suggestion_id=suggestion_id)
+    row = visible.first()
+    if row is None:
+        raise SuggestionNotFoundError(suggestion_id)
+    return row, access
+
+
+def _to_contract(row: WarehouseSuggestion, access: SubjectAccess) -> Suggestion:
+    return Suggestion(
+        id=row.id,
+        kind=WarehouseSuggestionKind(row.kind),
+        subject_kind=WarehouseSuggestionSubjectKind(row.subject_kind),
+        subject_id=row.subject_id,
+        payload=row.payload,
+        payload_version=row.payload_version,
+        evidence=row.evidence,
+        evidence_window_start=row.evidence_window_start,
+        evidence_window_end=row.evidence_window_end,
+        last_seen_at=row.last_seen_at,
+        score=row.score,
+        status=WarehouseSuggestionStatus(row.status),
+        surfaced_at=row.surfaced_at,
+        reviewed_by=_to_reviewer(row),
+        reviewed_at=row.reviewed_at,
+        dismissal_reason=WarehouseSuggestionDismissalReason(row.dismissal_reason) if row.dismissal_reason else None,
+        dismissal_note=row.dismissal_note,
+        created_asset=row.created_asset,
+        asset_outcome=WarehouseSuggestionAssetOutcome(row.asset_outcome) if row.asset_outcome else None,
+        can_act=access.can_act_on(row),
+    )
+
+
+def _to_reviewer(row: WarehouseSuggestion) -> SuggestionReviewer | None:
+    if row.reviewed_by is None:
+        return None
+    return SuggestionReviewer(id=row.reviewed_by.id, first_name=row.reviewed_by.first_name, email=row.reviewed_by.email)

products/warehouse_suggestions/backend/facade/contracts.py

@@ -0,0 +1,84 @@
+"""Contract types for warehouse_suggestions."""
+
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+from posthog.dataclasses import frozen
+
+from .enums import (
+    WarehouseSuggestionAssetOutcome,
+    WarehouseSuggestionDismissalReason,
+    WarehouseSuggestionKind,
+    WarehouseSuggestionStatus,
+    WarehouseSuggestionSubjectKind,
+)
+
+
+@frozen
+class SuggestionDraft:
+    kind: WarehouseSuggestionKind
+    fingerprint: str
+    subject_kind: WarehouseSuggestionSubjectKind
+    subject_id: UUID
+    payload: dict[str, Any]
+    payload_version: int
+    rules_version: str
+    evidence: dict[str, Any]
+    evidence_window_start: datetime
+    evidence_window_end: datetime
+    score: float
+    score_inputs: dict[str, Any]
+    run_id: str
+
+
+@frozen
+class SuggestionReviewer:
+    id: int
+    first_name: str
+    email: str
+
+
+@frozen
+class Suggestion:
+    id: UUID
+    kind: WarehouseSuggestionKind
+    subject_kind: WarehouseSuggestionSubjectKind
+    subject_id: UUID
+    payload: dict[str, Any]
+    payload_version: int
+    evidence: dict[str, Any]
+    evidence_window_start: datetime
+    evidence_window_end: datetime
+    last_seen_at: datetime
+    score: float
+    status: WarehouseSuggestionStatus
+    surfaced_at: datetime | None
+    reviewed_by: SuggestionReviewer | None
+    reviewed_at: datetime | None
+    dismissal_reason: WarehouseSuggestionDismissalReason | None
+    dismissal_note: str | None
+    created_asset: dict[str, Any] | None
+    asset_outcome: WarehouseSuggestionAssetOutcome | None
+    can_act: bool
+
+
+@frozen
+class SuggestionPage:
+    count: int
+    results: list[Suggestion]
+
+
+class SuggestionNotFoundError(Exception):
+    pass
+
+
+class SubjectEditAccessRequiredError(Exception):
+    def __init__(self, subject_kind: WarehouseSuggestionSubjectKind) -> None:
+        super().__init__(subject_kind)
+        self.subject_kind = subject_kind
+
+
+class SuggestionAlreadyDecidedError(Exception):
+    def __init__(self, current: WarehouseSuggestionStatus, requested: WarehouseSuggestionStatus) -> None:
+        super().__init__(f"This suggestion is {current.label.lower()} and cannot become {requested.label.lower()}.")