feat: enforce competency-hierarchy dominance for criteria

src/openedx_learning/applets/cbe/api.py

@@ -7,7 +7,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from decimal import Decimal
-from typing import Any, Sequence
+from typing import Any, Iterable, Sequence
 
 from django.core.exceptions import NON_FIELD_ERRORS, PermissionDenied, ValidationError
 from django.db import connection, transaction
@@ -19,12 +19,13 @@
 
 from openedx_catalog.api import get_course_run
 from openedx_catalog.models_api import CourseRun
-from openedx_tagging.api import create_taxonomy, get_object_tags, tag_object
+from openedx_tagging.api import create_taxonomy, get_ancestor_tags, get_descendant_tags, get_object_tags, tag_object
 from openedx_tagging.models import ObjectTag, Tag, Taxonomy
 from openedx_tagging.rules import ObjectTagPermissionItem, UserType
 
 from .data import GradedObjectScore
 from .models import (
+    CompetencyCourseRun,
     CompetencyCriteriaGroup,
     CompetencyCriterion,
     CompetencyRuleProfile,
@@ -45,6 +46,7 @@
     "create_competency_taxonomy",
     "get_competency_criteria_tree",
     "get_competency_rule_profiles",
+    "get_courses_with_relative_criteria",
     "is_competency_taxonomy",
     "record_graded_object_statuses",
     "resolve_competency_tag",
@@ -260,6 +262,58 @@ def resolve_supplied_leaf_group(group_id: int, tag: Tag, course_run: CourseRun)
     return group
 
 
+def get_courses_with_relative_criteria(
+    tag_ids: Iterable[int], course: CourseRun | None = None,
+) -> dict[CourseRun, Tag]:
+    """
+    Return every course already blocked for one of ``tag_ids`` by an existing CompetencyCriterion.
+
+    Maps each conflicting CourseRun to the lowest-id tag (of ``tag_ids``) with a criterion there,
+    narrowed to ``course`` when given. Archived criteria and criteria under an archived group are ignored.
+    """
+    groups: QuerySet[CompetencyCriteriaGroup, CompetencyCriteriaGroup] = CompetencyCriteriaGroup.objects.filter(
+        tag_id__in=tag_ids,
+        archived=False,
+        parent__archived=False,
+        parent__parent__archived=False,
+        # Only a leaf's parent carries a course, so this skips criteria misplaced on other groups.
+        parent__course__isnull=False,
+        criteria__archived=False,
+    )
+    if course is not None:
+        groups = groups.filter(parent__course=course)
+    result: dict[CourseRun, Tag] = {}
+    for group in groups.select_related("tag", "parent__course").distinct().order_by("tag_id"):
+        assert group.parent is not None, f"CompetencyCriteriaGroup {group.id} has no parent"
+        assert group.parent.course is not None, f"CompetencyCriteriaGroup {group.id}'s parent has no course"
+        result.setdefault(group.parent.course, group.tag)
+    return result
+
+
+def _validate_containment(group: CompetencyCriteriaGroup, course: CourseRun) -> None:
+    """
+    Raise ValidationError (keyed "tag_id") if ``group.tag`` conflicts with an existing criterion.
+
+    A conflict is a taxonomy ancestor or descendant of ``group.tag`` that already has a
+    CompetencyCriterion in ``course``.
+    """
+    tag = group.tag
+    relative_ids = (
+        set(get_ancestor_tags(tag).values_list("id", flat=True))
+        | set(get_descendant_tags(tag).values_list("id", flat=True))
+    )
+    if not relative_ids:
+        return
+    conflicting_tag = get_courses_with_relative_criteria(relative_ids, course=course).get(course)
+    if conflicting_tag is not None:
+        raise ValidationError({
+            "tag_id": _(
+                "Tag '{tag_value}' already has a criterion in course {course_key}, and is a taxonomy "
+                "ancestor or descendant of this criterion's tag."
+            ).format(tag_value=conflicting_tag.value, course_key=course.course_key),
+        })
+
+
 def _get_applicable_rule_profile(taxonomy_id: int) -> CompetencyRuleProfile | None:
     """
     Return the live profile ADR-0002 Decision 4 assigns a criterion in ``taxonomy_id``, or None if there is none.
@@ -281,6 +335,8 @@ def _get_applicable_rule_profile(taxonomy_id: int) -> CompetencyRuleProfile | No
 def create_competency_criterion(
     group: CompetencyCriteriaGroup,
     object_id: str,
+    course_run: CourseRun,
+    *,
     rule_profile_id: int | None = None,
     rule_type_override: str | None = None,
     rule_payload_override: dict | None = None,
@@ -300,10 +356,16 @@ def create_competency_criterion(
     else the live system default, rather than persisting null: ADR-0002 Decision 4 always
     resolves to a concrete profile, and the model's own xor-override constraint forbids leaving
     all three fields null.
+
+    ``course_run`` is ``group``'s course. Must run inside a transaction holding the tag and
+    course run locks that associate_competency_criterion() takes, so the dominance check below
+    can't race another create.
     """
     tag = group.tag
     # tag.taxonomy_id is nullable at the model level; already guaranteed set by the caller.
     assert tag.taxonomy_id is not None
+    _validate_containment(group, course_run)
+
     existing_values = [existing_tag.value for existing_tag in get_object_tags(object_id, taxonomy_id=tag.taxonomy_id)]
     if tag.value not in existing_values:
         existing_values.append(tag.value)
@@ -319,8 +381,6 @@ def create_competency_criterion(
             raise CompetencyRuleProfile.DoesNotExist("No live rule profile applies to this criterion.")
         rule_profile_id = profile.id
 
-    # Parent-competency dominance/containment validation belongs here, before creation.
-
     return CompetencyCriterion.objects.create(
         group=group,
         object_tag=object_tag,
@@ -433,8 +493,14 @@ def associate_competency_criterion(
     _validate_rule_fields(rule_profile_id, rule_type_override, rule_payload_override)
 
     with transaction.atomic():
+        # Lock order is always the tag, then the course run, so two creates can't deadlock.
         # Serializes the duplicate check and root creation; MySQL has no partial unique index for one root per tag.
         Tag.objects.select_for_update().get(pk=tag.pk)
+        # Serializes the per-course dominance check on CBE's own row, not the catalog's CourseRun.
+        # Created first, since a locking read of a row that doesn't exist yet locks nothing; get_or_create()
+        # resolves two first creates racing on the primary key.
+        CompetencyCourseRun.objects.get_or_create(course_run=course_run)
+        CompetencyCourseRun.objects.select_for_update().get(pk=course_run.pk)
         _reject_duplicate_criterion(tag, object_id, group_id)
         if group_id is not None:
             group = resolve_supplied_leaf_group(group_id, tag, course_run)
@@ -443,6 +509,7 @@ def associate_competency_criterion(
         return create_competency_criterion(
             group=group,
             object_id=object_id,
+            course_run=course_run,
             rule_profile_id=rule_profile_id,
             rule_type_override=rule_type_override,
             rule_payload_override=rule_payload_override,

src/openedx_learning/applets/cbe/models/__init__.py

@@ -4,6 +4,7 @@
 
 from ..rule_payloads import RuleType
 from .competency_taxonomy import CompetencyTaxonomy
+from .course_run import CompetencyCourseRun
 from .criteria import CompetencyCriteriaGroup, CompetencyCriterion, CompetencyRuleProfile, LogicOperator
 from .learner_status import (
     CompetencyMasteryStatus,
@@ -14,6 +15,7 @@
 )
 
 __all__ = [
+    "CompetencyCourseRun",
     "CompetencyCriteriaGroup",
     "CompetencyCriterion",
     "CompetencyMasteryStatus",

src/openedx_learning/applets/cbe/models/course_run.py

@@ -0,0 +1,32 @@
+"""
+CompetencyCourseRun: CBE's own per-course row for a catalog CourseRun.
+"""
+from __future__ import annotations
+
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from openedx_catalog.models_api import CourseRun
+
+__all__ = [
+    "CompetencyCourseRun",
+]
+
+
+class CompetencyCourseRun(models.Model):
+    """
+    A course run that has competency criteria, created with its first criterion.
+
+    Criterion creation locks this row, rather than the catalog's CourseRun, to serialize the
+    per-course dominance check without blocking other apps' writes that reference the course run.
+
+    .. no_pii:
+    """
+
+    course_run = models.OneToOneField(
+        CourseRun,
+        primary_key=True,
+        on_delete=models.CASCADE,
+        related_name="competency_course_run",
+        help_text=_("The catalog course run this row extends."),
+    )

src/openedx_learning/migrations/0016_add_competency_course_run.py

@@ -0,0 +1,21 @@
+# Generated by Django 5.2.17 on 2026-10-20 12:42
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('openedx_catalog', '0001_initial'),
+        ('openedx_learning', '0015_criteria_group_unique_constraints'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CompetencyCourseRun',
+            fields=[
+                ('course_run', models.OneToOneField(help_text='The catalog course run this row extends.', on_delete=django.db.models.deletion.CASCADE, primary_key=True, related_name='competency_course_run', serialize=False, to='openedx_catalog.courserun')),
+            ],
+        ),
+    ]

src/openedx_tagging/api.py

@@ -169,6 +169,30 @@ def get_children_tags(
     return taxonomy.get_filtered_tags(parent_tag_value=parent_tag_value, depth=1)
 
 
+def get_ancestor_tags(tag: Tag) -> QuerySet[Tag]:
+    """
+    Return every ancestor of ``tag`` (its parent, grandparent, ... up to the root).
+
+    Returns an empty queryset if ``tag``'s taxonomy is unset or allows free text, since
+    free-text tags have no taxonomy-defined hierarchy.
+    """
+    if tag.taxonomy is None or tag.taxonomy.allow_free_text:
+        return Tag.objects.none()
+    return Tag.objects.filter(taxonomy=tag.taxonomy, value__in=tag.get_lineage()[:-1])
+
+
+def get_descendant_tags(tag: Tag) -> QuerySet[Tag]:
+    """
+    Return every descendant of ``tag`` at any depth, excluding ``tag`` itself.
+
+    Returns an empty queryset if ``tag``'s taxonomy is unset or allows free text, since
+    free-text tags have no taxonomy-defined hierarchy.
+    """
+    if tag.taxonomy is None or tag.taxonomy.allow_free_text:
+        return Tag.objects.none()
+    return Tag.objects.filter(taxonomy=tag.taxonomy, depth__gt=tag.depth, lineage__startswith=tag.lineage)
+
+
 def resync_object_tags(object_tags: QuerySet | None = None) -> int:
     """
     Reconciles ObjectTag entries with any changes made to their associated taxonomies and tags.