Classification - run condition text automatch as an async task, batched per import #1780

annotation/tests/test_annotation_disk_space.py

@@ -49,6 +49,20 @@
 STANDARD = VariantAnnotationPipelineType.STANDARD
 
 
+def past_vep_kwargs() -> dict:
+    """ Timestamps/counts a run carries once VEP has finished, ie ANNOTATION_COMPLETED - what the import
+        lane checks before it does anything. """
+    now = timezone.now()
+    return {
+        "count": 100,
+        "dump_start": now - timedelta(minutes=5),
+        "dump_end": now - timedelta(minutes=4),
+        "dump_count": 100,
+        "annotation_start": now - timedelta(minutes=4),
+        "annotation_end": now,
+    }
+
+
 @override_settings(**get_fake_annotation_settings_dict(columns_version=2))
 class AnnotationRunCleanupTestCase(TestCase):
     """ Each way into the cleanup module, and the one case that must not reach it. """
@@ -134,7 +148,8 @@ def test_failed_import_keeps_everything(self):
         with tempfile.TemporaryDirectory() as tmp_dir, \
                 override_settings(ANNOTATION_VCF_DUMP_DIR=tmp_dir,
                                   ANNOTATION_DELETE_TEMP_FILES_ON_SUCCESS=True):
-            run, paths, _ = self._run_with_output(tmp_dir)
+            run, paths, _ = self._run_with_output(tmp_dir, **past_vep_kwargs())
+            self.assertEqual(run.status, AnnotationStatus.ANNOTATION_COMPLETED)
 
             with mock.patch.object(VEPRunner, "import_results",
                             side_effect=RuntimeError("import blew up")), \
@@ -204,12 +219,8 @@ def _make_run(self, lo_idx, hi_idx, status=AnnotationStatus.CREATED):
         # count stamped as the count lane would have, so the run is ready for the run lanes
         run = AnnotationRun.objects.create(annotation_range_lock=lock, pipeline_type=STANDARD, count=100)
         if status == AnnotationStatus.ANNOTATION_COMPLETED:
-            now = timezone.now()  # past VEP, waiting on the import lane
-            run.dump_start = now - timedelta(minutes=5)
-            run.dump_end = now - timedelta(minutes=4)
-            run.dump_count = 100
-            run.annotation_start = now - timedelta(minutes=4)
-            run.annotation_end = now
+            for k, v in past_vep_kwargs().items():  # past VEP, waiting on the import lane
+                setattr(run, k, v)
             run.vcf_annotated_filename = "/does/not/need/to/exist.vcf.gz"
             run.save()
             self.assertEqual(run.status, AnnotationStatus.ANNOTATION_COMPLETED)

classification/migrations/0176_conditiontext_pending_automatch.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.1 on 2026-08-26 06:38
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('classification', '0175_ekey_gene_fusion_options'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='conditiontext',
+            name='pending_automatch',
+            field=models.BooleanField(default=False),
+        ),
+    ]

classification/models/condition_text_matching.py

@@ -69,6 +69,9 @@ class ConditionText(TimeStampedModel, GuardianPermissionsMixin):
 
     classifications_count = models.IntegerField(default=0)
     classifications_count_outstanding = models.IntegerField(default=0)
+    # set when a new root/gene level appears, drained by the condition_text_automatch_task beat sweep -
+    # automatching can call the external Monarch search API so it can't run in the publishing request (#1780)
+    pending_automatch = models.BooleanField(default=False)
 
     class Meta:
         unique_together = ("normalized_text", "lab")
@@ -441,13 +444,15 @@ def sync_condition_text_classification(cm: ClassificationModification, update_co
                 debug_timer.tick("Condition Text Matching - create new entry")
 
             if attempt_automatch and (new_root or new_gene_level):
-                ConditionTextMatch.attempt_automatch(ct, gene_symbol=gene_symbol)
-                debug_timer.tick("Condition Text Matching - auto match")
-            elif update_counts:
+                ct.pending_automatch = True
+
+            if update_counts:
                 ct.classifications_count += 1
                 is_valid = root.is_valid or gene_level.is_valid or mode_of_inheritance_level.is_valid or (existing and existing.is_valid)
                 if not is_valid:
                     ct.classifications_count_outstanding += 1
+
+            if update_counts or ct.pending_automatch:
                 ct.save()
                 debug_timer.tick("Condition Text Matching - update count quick")
 

classification/tasks/condition_text_automatch_task.py

@@ -0,0 +1,22 @@
+import celery
+
+from classification.models.classification_import_run import ClassificationImportRun
+from classification.models.condition_text_matching import ConditionText, ConditionTextMatch
+
+
+@celery.shared_task
+def condition_text_automatch_task():
+    """
+    Celery beat sweep that automatches every ConditionText flagged pending_automatch. Automatching
+    can call the external Monarch search API, which can be slow or down entirely, so it runs here
+    rather than in the publishing request (#1780). While a bulk import is ongoing the sweep stands
+    aside - the flags accumulate and the first sweep after completion handles the whole batch,
+    deduped to one automatch per distinct condition text. The flags are also the crash recovery:
+    anything a dead worker left behind is picked up by the next sweep.
+    """
+    if ClassificationImportRun.ongoing_imports():
+        return
+    for ct in ConditionText.objects.filter(pending_automatch=True).iterator():
+        # single-statement claim so overlapping sweeps can't automatch the same text twice
+        if ConditionText.objects.filter(pk=ct.pk, pending_automatch=True).update(pending_automatch=False):
+            ConditionTextMatch.attempt_automatch(condition_text=ct)

classification/tests/models/test_condition_text_automatch.py

@@ -0,0 +1,51 @@
+from unittest.mock import patch
+
+from django.test import TestCase
+from threadlocals.threadlocals import set_thread_variable
+
+from classification.models.classification_import_run import ClassificationImportRun
+from classification.models.condition_text_matching import ConditionText, ConditionTextMatch
+from classification.tasks.condition_text_automatch_task import condition_text_automatch_task
+from snpdb.models import Country, Lab, Organization
+
+
+class ConditionTextAutomatchTest(TestCase):
+
+    def setUp(self):
+        # NotificationBuilder logs an Event against the thread-local user - a client request in
+        # an earlier test leaves a rolled-back one behind
+        set_thread_variable('request', None)
+        org = Organization.objects.create(name='InstX', group_name='instx')
+        country = Country.objects.get_or_create(name='CountryA')[0]
+        self.lab = Lab.objects.create(name='Labby', organization=org, city='CityA',
+                                      country=country, group_name='instx/labby')
+
+    def _condition_text(self, text: str, pending: bool) -> ConditionText:
+        return ConditionText.objects.create(normalized_text=text, lab=self.lab, pending_automatch=pending)
+
+    @patch.object(ConditionTextMatch, 'attempt_automatch')
+    def test_sweep_automatches_each_pending_text_once(self, mock_automatch):
+        pending_1 = self._condition_text("condition 1", pending=True)
+        pending_2 = self._condition_text("condition 2", pending=True)
+        self._condition_text("condition 3", pending=False)
+
+        condition_text_automatch_task()
+
+        automatched = {call.kwargs["condition_text"].pk for call in mock_automatch.call_args_list}
+        self.assertEqual(automatched, {pending_1.pk, pending_2.pk})
+        self.assertFalse(ConditionText.objects.filter(pending_automatch=True).exists())
+
+    @patch.object(ConditionTextMatch, 'attempt_automatch')
+    def test_sweep_waits_for_ongoing_import(self, mock_automatch):
+        self._condition_text("condition 1", pending=True)
+        ClassificationImportRun.record_classification_import(identifier="test-import")
+
+        condition_text_automatch_task()
+        mock_automatch.assert_not_called()
+        self.assertTrue(ConditionText.objects.filter(pending_automatch=True).exists())
+
+        ClassificationImportRun.record_classification_import(identifier="test-import", is_complete=True)
+
+        condition_text_automatch_task()
+        mock_automatch.assert_called_once()
+        self.assertFalse(ConditionText.objects.filter(pending_automatch=True).exists())

variantgrid/celery.py

@@ -51,6 +51,14 @@
         'schedule': HOUR_SECS,  # Check every hour, only update if hash changed
     }
 
+# Condition text automatch sweep (#1780): publishing flags ConditionTexts (pending_automatch)
+# instead of calling the external Monarch search API in the request; this drains the flags, and
+# skips while a bulk import is ongoing so a sync's worth is handled as one deduped batch afterwards.
+app.conf.beat_schedule['condition-text-automatch'] = {
+    'task': 'classification.tasks.condition_text_automatch_task.condition_text_automatch_task',
+    'schedule': MINUTE_SECS * 5,
+}
+
 # Reclassification timelines (issue #1523): the analytics page builds what it can in the request,
 # this picks up anything left over, e.g. the morning after a large sync.
 app.conf.beat_schedule['reclassification-events-update'] = {

variantgrid/settings/components/celery_settings.py

@@ -136,6 +136,7 @@
     'classification.tasks.classification_import_process_variants_task',
     'classification.tasks.classification_import_task',
     'classification.tasks.classification_candidate_search_tasks',
+    'classification.tasks.condition_text_automatch_task',
     'genes.tasks.gene_coverage_tasks',
     'patients.tasks.extraction_matching_tasks',
     'pedigree.models',