feat(learnings): added learnings business logic

src/backend/apps/learnings/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class LearningsConfig(AppConfig):
+    name = "apps.learnings"
+    label = "learnings"

src/backend/apps/learnings/migrations/0001_initial.py

@@ -0,0 +1,64 @@
+# Generated by Django 5.2.11 on 2026-02-22 16:19
+
+import django.contrib.postgres.indexes
+import django.contrib.postgres.search
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('experiments', '0002_initial'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Learning',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('hypothesis', models.TextField(verbose_name='hypothesis')),
+                ('findings', models.TextField(verbose_name='findings')),
+                ('tags', models.JSONField(blank=True, default=list, verbose_name='tags')),
+                ('context_summary', models.TextField(blank=True, verbose_name='context summary')),
+                ('search_vector', django.contrib.postgres.search.SearchVectorField(blank=True, null=True, verbose_name='search vector')),
+                ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, verbose_name='updated at')),
+                ('created_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='created_learnings', to=settings.AUTH_USER_MODEL, verbose_name='created by')),
+                ('experiment', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='learning', to='experiments.experiment', verbose_name='experiment')),
+            ],
+            options={
+                'verbose_name': 'learning',
+                'verbose_name_plural': 'learnings',
+                'ordering': ['-created_at'],
+            },
+        ),
+        migrations.CreateModel(
+            name='LearningEdit',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('changes', models.JSONField(default=dict, verbose_name='changes')),
+                ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
+                ('edited_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='learning_edits', to=settings.AUTH_USER_MODEL, verbose_name='edited by')),
+                ('learning', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='edits', to='learnings.learning', verbose_name='learning')),
+            ],
+            options={
+                'verbose_name': 'learning edit',
+                'verbose_name_plural': 'learning edits',
+                'ordering': ['-created_at'],
+            },
+        ),
+        migrations.AddIndex(
+            model_name='learning',
+            index=django.contrib.postgres.indexes.GinIndex(fields=['search_vector'], name='idx_learning_search'),
+        ),
+        migrations.AddIndex(
+            model_name='learning',
+            index=models.Index(fields=['experiment'], name='idx_learning_experiment'),
+        ),
+    ]

src/backend/apps/learnings/models.py

@@ -0,0 +1,102 @@
+from typing import override
+
+from django.conf import settings
+from django.contrib.postgres.indexes import GinIndex
+from django.contrib.postgres.search import SearchVectorField
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from apps.core.models import BaseModel
+
+
+class Learning(BaseModel):
+    experiment = models.OneToOneField(
+        "experiments.Experiment",
+        on_delete=models.CASCADE,
+        related_name="learning",
+        verbose_name=_("experiment"),
+    )
+    hypothesis = models.TextField(
+        verbose_name=_("hypothesis"),
+    )
+    findings = models.TextField(
+        verbose_name=_("findings"),
+    )
+    tags = models.JSONField(
+        default=list,
+        blank=True,
+        verbose_name=_("tags"),
+    )
+    context_summary = models.TextField(
+        blank=True,
+        verbose_name=_("context summary"),
+    )
+    search_vector = SearchVectorField(
+        null=True,
+        blank=True,
+        verbose_name=_("search vector"),
+    )
+    created_by = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="created_learnings",
+        verbose_name=_("created by"),
+    )
+    created_at = models.DateTimeField(
+        auto_now_add=True,
+        verbose_name=_("created at"),
+    )
+    updated_at = models.DateTimeField(
+        auto_now=True,
+        verbose_name=_("updated at"),
+    )
+
+    class Meta:
+        verbose_name = _("learning")
+        verbose_name_plural = _("learnings")
+        ordering = ["-created_at"]
+        indexes = [
+            GinIndex(fields=["search_vector"], name="idx_learning_search"),
+            models.Index(fields=["experiment"], name="idx_learning_experiment"),
+        ]
+
+    @override
+    def __str__(self) -> str:
+        return f"Learning: {self.experiment.name}"
+
+
+class LearningEdit(BaseModel):
+    learning = models.ForeignKey(
+        Learning,
+        on_delete=models.CASCADE,
+        related_name="edits",
+        verbose_name=_("learning"),
+    )
+    edited_by = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="learning_edits",
+        verbose_name=_("edited by"),
+    )
+    changes = models.JSONField(
+        default=dict,
+        verbose_name=_("changes"),
+    )
+    created_at = models.DateTimeField(
+        auto_now_add=True,
+        verbose_name=_("created at"),
+    )
+
+    class Meta:
+        verbose_name = _("learning edit")
+        verbose_name_plural = _("learning edits")
+        ordering = ["-created_at"]
+
+    @override
+    def __str__(self) -> str:
+        editor = self.edited_by.username if self.edited_by else "system"
+        return f"Edit by {editor} on {self.learning}"

src/backend/apps/learnings/services.py

@@ -0,0 +1,268 @@
+import logging
+from typing import Any
+from uuid import UUID
+
+from django.db import connection, transaction
+from django.db.models import Q, QuerySet
+
+from apps.experiments.models import Experiment, ExperimentOutcome, ExperimentStatus
+from apps.guardrails.models import GuardrailTrigger
+from apps.learnings.models import Learning, LearningEdit
+from apps.metrics.models import ExperimentMetric
+from apps.users.models import User
+
+logger = logging.getLogger("lotty")
+
+
+def _is_postgres() -> bool:
+    return connection.vendor == "postgresql"
+
+
+@transaction.atomic
+def learning_create(
+    *,
+    experiment: Experiment,
+    hypothesis: str,
+    findings: str,
+    tags: list[str] | None = None,
+    context_summary: str = "",
+    user: User | None = None,
+) -> Learning:
+    learning = Learning(
+        experiment=experiment,
+        hypothesis=hypothesis,
+        findings=findings,
+        tags=tags or [],
+        context_summary=context_summary,
+        created_by=user,
+    )
+    learning.save()
+    _update_search_vector(learning)
+    return learning
+
+
+def learning_update(
+    *,
+    learning: Learning,
+    user: User | None = None,
+    **fields: Any,
+) -> Learning:
+    allowed = {"hypothesis", "findings", "tags", "context_summary"}
+    changes: dict[str, Any] = {}
+    for key in fields:
+        if key not in allowed:
+            raise ValueError(f"Field '{key}' cannot be updated.")
+    for key, value in fields.items():
+        if value is not None:
+            old_value = getattr(learning, key)
+            if old_value != value:
+                changes[key] = {"old": old_value, "new": value}
+                setattr(learning, key, value)
+    if changes:
+        learning.save()
+        _update_search_vector(learning)
+        LearningEdit.objects.create(
+            learning=learning,
+            edited_by=user,
+            changes=changes,
+        )
+    return learning
+
+
+def learning_delete(*, learning: Learning) -> None:
+    learning.delete()
+
+
+def learning_get(learning_id: UUID) -> Learning | None:
+    try:
+        return Learning.objects.select_related(
+            "experiment__flag",
+            "experiment__owner",
+            "created_by",
+        ).get(pk=learning_id)
+    except Learning.DoesNotExist:
+        return None
+
+
+def learning_list(
+    *,
+    flag_id: UUID | None = None,
+    owner_id: UUID | None = None,
+    outcome: str | None = None,
+    tag: str | None = None,
+    search: str | None = None,
+) -> QuerySet[Learning]:
+    qs = Learning.objects.select_related(
+        "experiment__flag",
+        "experiment__owner",
+        "created_by",
+    ).all()
+
+    if flag_id is not None:
+        qs = qs.filter(experiment__flag_id=flag_id)
+    if owner_id is not None:
+        qs = qs.filter(experiment__owner_id=owner_id)
+    if outcome is not None:
+        qs = qs.filter(experiment__outcome__outcome=outcome)
+    if tag is not None:
+        if _is_postgres():
+            qs = qs.filter(tags__contains=[tag])
+        else:
+            qs = qs.filter(tags__icontains=tag)
+    if search is not None:
+        if _is_postgres():
+            from django.contrib.postgres.search import SearchQuery, SearchRank
+            query = SearchQuery(search, config="english")
+            qs = qs.filter(search_vector=query).annotate(
+                rank=SearchRank("search_vector", query)
+            ).order_by("-rank")
+        else:
+            qs = qs.filter(
+                Q(hypothesis__icontains=search)
+                | Q(findings__icontains=search)
+                | Q(context_summary__icontains=search)
+            )
+
+    return qs
+
+
+def learning_edit_list(learning_id: UUID) -> QuerySet[LearningEdit]:
+    return LearningEdit.objects.select_related(
+        "edited_by",
+    ).filter(learning_id=learning_id)
+
+
+def find_similar_learnings(
+    *,
+    experiment_id: UUID,
+    limit: int = 5,
+) -> list[dict[str, Any]]:
+    try:
+        experiment = Experiment.objects.select_related("flag").get(pk=experiment_id)
+    except Experiment.DoesNotExist:
+        return []
+
+    experiment_flag_id = experiment.flag_id
+    experiment_metric_ids = set(
+        ExperimentMetric.objects.filter(
+            experiment_id=experiment_id
+        ).values_list("metric_id", flat=True)
+    )
+    experiment_tags = set()
+    try:
+        learning = Learning.objects.get(experiment_id=experiment_id)
+        experiment_tags = set(learning.tags or [])
+    except Learning.DoesNotExist:
+        pass
+
+    had_guardrail_triggers = GuardrailTrigger.objects.filter(
+        experiment_id=experiment_id
+    ).exists()
+
+    candidates = Learning.objects.select_related(
+        "experiment__flag",
+        "experiment__owner",
+        "experiment__outcome",
+    ).exclude(
+        experiment_id=experiment_id,
+    ).order_by("-created_at")[:100]
+
+    scored: list[tuple[float, Learning]] = []
+    for candidate in candidates:
+        score = _jaccard_score(
+            candidate=candidate,
+            experiment_flag_id=experiment_flag_id,
+            experiment_metric_ids=experiment_metric_ids,
+            experiment_tags=experiment_tags,
+            had_guardrail_triggers=had_guardrail_triggers,
+        )
+        if score > 0:
+            scored.append((score, candidate))
+
+    scored.sort(key=lambda x: x[0], reverse=True)
+    results: list[dict[str, Any]] = []
+    for score, candidate in scored[:limit]:
+        outcome_data = None
+        if hasattr(candidate.experiment, "outcome"):
+            try:
+                exp_outcome = candidate.experiment.outcome
+                outcome_data = {
+                    "outcome": exp_outcome.outcome,
+                    "rationale": exp_outcome.rationale,
+                    "winning_variant": str(exp_outcome.winning_variant_id) if exp_outcome.winning_variant_id else None,
+                }
+            except ExperimentOutcome.DoesNotExist:
+                pass
+
+        trigger_count = GuardrailTrigger.objects.filter(
+            experiment_id=candidate.experiment_id
+        ).count()
+
+        results.append({
+            "learning_id": str(candidate.pk),
+            "experiment_id": str(candidate.experiment_id),
+            "experiment_name": candidate.experiment.name,
+            "flag_key": candidate.experiment.flag.key,
+            "hypothesis": candidate.hypothesis,
+            "findings": candidate.findings,
+            "tags": candidate.tags,
+            "outcome": outcome_data,
+            "had_guardrail_triggers": trigger_count > 0,
+            "guardrail_trigger_count": trigger_count,
+            "similarity_score": round(score, 3),
+        })
+
+    return results
+
+
+def _jaccard_score(
+    *,
+    candidate: Learning,
+    experiment_flag_id: UUID,
+    experiment_metric_ids: set,
+    experiment_tags: set,
+    had_guardrail_triggers: bool,
+) -> float:
+    score = 0.0
+
+    if candidate.experiment.flag_id == experiment_flag_id:
+        score += 0.4
+
+    candidate_tags = set(candidate.tags or [])
+    if experiment_tags and candidate_tags:
+        intersection = experiment_tags & candidate_tags
+        union = experiment_tags | candidate_tags
+        if union:
+            score += 0.3 * (len(intersection) / len(union))
+
+    candidate_metric_ids = set(
+        ExperimentMetric.objects.filter(
+            experiment_id=candidate.experiment_id,
+        ).values_list("metric_id", flat=True)
+    )
+    if experiment_metric_ids and candidate_metric_ids:
+        m_intersection = experiment_metric_ids & candidate_metric_ids
+        m_union = experiment_metric_ids | candidate_metric_ids
+        if m_union:
+            score += 0.2 * (len(m_intersection) / len(m_union))
+
+    candidate_had_triggers = GuardrailTrigger.objects.filter(
+        experiment_id=candidate.experiment_id
+    ).exists()
+    if had_guardrail_triggers == candidate_had_triggers:
+        score += 0.1
+
+    return score
+
+
+def _update_search_vector(learning: Learning) -> None:
+    if not _is_postgres():
+        return
+    from django.contrib.postgres.search import SearchVector
+    Learning.objects.filter(pk=learning.pk).update(
+        search_vector=(
+            SearchVector("hypothesis", weight="A", config="english")
+            + SearchVector("findings", weight="B", config="english")
+            + SearchVector("context_summary", weight="C", config="english")
+        )
+    )

src/backend/apps/learnings/tests/test_learnings.py

@@ -0,0 +1,279 @@
+from typing import override
+
+from django.test import TestCase
+
+from apps.experiments.models import ExperimentOutcome, ExperimentStatus, OutcomeType
+from apps.experiments.services import experiment_complete
+from apps.experiments.tests.helpers import add_two_variants, make_experiment, make_flag
+from apps.guardrails.models import Guardrail, GuardrailTrigger
+from apps.learnings.models import Learning, LearningEdit
+from apps.learnings.services import (
+    find_similar_learnings,
+    learning_create,
+    learning_delete,
+    learning_edit_list,
+    learning_get,
+    learning_list,
+    learning_update,
+)
+from apps.metrics.models import ExperimentMetric, MetricDefinition, MetricType
+from apps.reviews.tests.helpers import make_admin
+from apps.users.tests.helpers import make_user
+
+
+class LearningCRUDTest(TestCase):
+    @override
+    def setUp(self) -> None:
+        self.user = make_admin("_lcrud")
+        self.exp = make_experiment(suffix="_lcrud", owner=self.user)
+
+    def test_create_learning(self) -> None:
+        l = learning_create(
+            experiment=self.exp,
+            hypothesis="Changing button color increases CTR",
+            findings="Blue variant showed +5% CTR improvement",
+            tags=["ui", "ctr", "button"],
+            context_summary="Checkout page button color test",
+            user=self.user,
+        )
+        self.assertEqual(l.experiment, self.exp)
+        self.assertEqual(l.hypothesis, "Changing button color increases CTR")
+        self.assertEqual(l.findings, "Blue variant showed +5% CTR improvement")
+        self.assertEqual(l.tags, ["ui", "ctr", "button"])
+        self.assertEqual(l.created_by, self.user)
+
+    def test_create_learning_minimal(self) -> None:
+        l = learning_create(
+            experiment=self.exp,
+            hypothesis="Test hypothesis",
+            findings="Test findings",
+        )
+        self.assertEqual(l.tags, [])
+        self.assertEqual(l.context_summary, "")
+        self.assertIsNone(l.created_by)
+
+    def test_create_duplicate_learning_fails(self) -> None:
+        learning_create(
+            experiment=self.exp,
+            hypothesis="First",
+            findings="First findings",
+        )
+        with self.assertRaises(Exception):
+            learning_create(
+                experiment=self.exp,
+                hypothesis="Second",
+                findings="Second findings",
+            )
+
+    def test_get_learning(self) -> None:
+        l = learning_create(
+            experiment=self.exp,
+            hypothesis="Test",
+            findings="Test findings",
+        )
+        fetched = learning_get(l.pk)
+        self.assertIsNotNone(fetched)
+        self.assertEqual(fetched.pk, l.pk)
+
+    def test_get_nonexistent_learning(self) -> None:
+        import uuid
+        result = learning_get(uuid.uuid4())
+        self.assertIsNone(result)
+
+    def test_delete_learning(self) -> None:
+        l = learning_create(
+            experiment=self.exp,
+            hypothesis="Test",
+            findings="Test findings",
+        )
+        learning_delete(learning=l)
+        self.assertIsNone(learning_get(l.pk))
+
+
+class LearningUpdateTest(TestCase):
+    @override
+    def setUp(self) -> None:
+        self.user = make_admin("_lupd")
+        self.exp = make_experiment(suffix="_lupd", owner=self.user)
+        self.learning = learning_create(
+            experiment=self.exp,
+            hypothesis="Original hypothesis",
+            findings="Original findings",
+            tags=["original"],
+            user=self.user,
+        )
+
+    def test_update_hypothesis(self) -> None:
+        editor = make_user(
+            username="editor_lupd",
+            email="editor_lupd@lotty.local",
+        )
+        l = learning_update(
+            learning=self.learning,
+            user=editor,
+            hypothesis="Updated hypothesis",
+        )
+        self.assertEqual(l.hypothesis, "Updated hypothesis")
+
+    def test_update_creates_audit_trail(self) -> None:
+        editor = make_user(
+            username="editor2_lupd",
+            email="editor2_lupd@lotty.local",
+        )
+        learning_update(
+            learning=self.learning,
+            user=editor,
+            findings="New findings",
+        )
+        edits = learning_edit_list(self.learning.pk)
+        self.assertEqual(edits.count(), 1)
+        edit = edits.first()
+        self.assertEqual(edit.edited_by, editor)
+        self.assertIn("findings", edit.changes)
+        self.assertEqual(edit.changes["findings"]["old"], "Original findings")
+        self.assertEqual(edit.changes["findings"]["new"], "New findings")
+
+    def test_update_no_change_no_audit(self) -> None:
+        learning_update(
+            learning=self.learning,
+            user=self.user,
+            hypothesis="Original hypothesis",
+        )
+        edits = learning_edit_list(self.learning.pk)
+        self.assertEqual(edits.count(), 0)
+
+    def test_update_disallowed_field(self) -> None:
+        with self.assertRaises(ValueError):
+            learning_update(
+                learning=self.learning,
+                user=self.user,
+                experiment="other",
+            )
+
+    def test_update_multiple_fields(self) -> None:
+        learning_update(
+            learning=self.learning,
+            user=self.user,
+            hypothesis="New hyp",
+            findings="New find",
+            tags=["new", "tags"],
+        )
+        edits = learning_edit_list(self.learning.pk)
+        self.assertEqual(edits.count(), 1)
+        edit = edits.first()
+        self.assertIn("hypothesis", edit.changes)
+        self.assertIn("findings", edit.changes)
+        self.assertIn("tags", edit.changes)
+
+
+class LearningListTest(TestCase):
+    @override
+    def setUp(self) -> None:
+        self.user = make_admin("_llist")
+        self.flag = make_flag(suffix="_llist")
+        self.exp = make_experiment(
+            flag=self.flag,
+            owner=self.user,
+            suffix="_llist",
+        )
+        self.learning = learning_create(
+            experiment=self.exp,
+            hypothesis="Test hypothesis about button color",
+            findings="Blue variant performed better",
+            tags=["ui", "button"],
+        )
+
+    def test_list_all(self) -> None:
+        learnings = learning_list()
+        self.assertGreaterEqual(learnings.count(), 1)
+
+    def test_filter_by_flag(self) -> None:
+        learnings = learning_list(flag_id=self.flag.pk)
+        self.assertEqual(learnings.count(), 1)
+
+    def test_filter_by_tag(self) -> None:
+        learnings = learning_list(tag="ui")
+        self.assertTrue(all("ui" in l.tags for l in learnings))
+
+    def test_filter_by_tag_no_match(self) -> None:
+        learnings = learning_list(tag="nonexistent")
+        self.assertEqual(learnings.count(), 0)
+
+    def test_filter_by_owner(self) -> None:
+        learnings = learning_list(owner_id=self.user.pk)
+        self.assertEqual(learnings.count(), 1)
+
+    def test_fulltext_search(self) -> None:
+        learnings = learning_list(search="button color")
+        self.assertGreaterEqual(learnings.count(), 1)
+
+
+class SimilarLearningsTest(TestCase):
+    @override
+    def setUp(self) -> None:
+        self.user = make_admin("_sim")
+        self.flag = make_flag(suffix="_sim1")
+
+        self.exp1 = make_experiment(flag=self.flag, owner=self.user, suffix="_sim1")
+        self.learning1 = learning_create(
+            experiment=self.exp1,
+            hypothesis="Test button color",
+            findings="Blue won",
+            tags=["ui", "button", "checkout"],
+        )
+
+        flag2 = make_flag(suffix="_sim2")
+        self.exp2 = make_experiment(flag=flag2, owner=self.user, suffix="_sim2")
+        self.learning2 = learning_create(
+            experiment=self.exp2,
+            hypothesis="Test font size",
+            findings="Larger font won",
+            tags=["ui", "font"],
+        )
+
+        self.exp3 = make_experiment(
+            flag=self.flag,
+            owner=self.user,
+            suffix="_sim3",
+            name="SameFlag Exp",
+        )
+        self.learning3 = learning_create(
+            experiment=self.exp3,
+            hypothesis="Another button test",
+            findings="Red won",
+            tags=["ui", "button", "checkout"],
+        )
+
+    def test_find_similar_same_flag(self) -> None:
+        results = find_similar_learnings(experiment_id=self.exp1.pk)
+        self.assertGreaterEqual(len(results), 1)
+        flag_keys = [r["flag_key"] for r in results]
+        self.assertIn(self.flag.key, flag_keys)
+
+    def test_find_similar_tag_overlap(self) -> None:
+        results = find_similar_learnings(experiment_id=self.exp1.pk)
+        self.assertTrue(len(results) > 0)
+        top_result = results[0]
+        self.assertGreater(top_result["similarity_score"], 0)
+
+    def test_find_similar_excludes_self(self) -> None:
+        results = find_similar_learnings(experiment_id=self.exp1.pk)
+        result_exp_ids = [r["experiment_id"] for r in results]
+        self.assertNotIn(str(self.exp1.pk), result_exp_ids)
+
+    def test_find_similar_nonexistent_experiment(self) -> None:
+        import uuid
+        results = find_similar_learnings(experiment_id=uuid.uuid4())
+        self.assertEqual(results, [])
+
+    def test_find_similar_respects_limit(self) -> None:
+        results = find_similar_learnings(experiment_id=self.exp1.pk, limit=1)
+        self.assertLessEqual(len(results), 1)
+
+    def test_similar_highest_score_is_same_flag_and_tags(self) -> None:
+        results = find_similar_learnings(experiment_id=self.exp1.pk)
+        if len(results) >= 2:
+            self.assertGreaterEqual(
+                results[0]["similarity_score"],
+                results[1]["similarity_score"],
+            )