Merge pull request #15051 from rtibblesbot/issue-15014-3d4576

Adopt supervisor ownership and reconciliation in the Android task layer

Author: rtibbles

.github/workflows/tox.yml

@@ -385,6 +385,35 @@ jobs:
       - name: Run tests
         working-directory: ${{ matrix.working-directory }}
         run: uv run python -O -m pytest
+  # The Android host tests exercise the WorkManager task layer against real
+  # Kolibri core (supervisor ownership, reconciliation), so they need Kolibri's
+  # Django test settings and the kolibri/dist pythonpath. They live under the
+  # Chaquopy source dir but can't use the member's own pytest config, so run
+  # them from the repo root with the root pytest.ini. Pinned to Python 3.10 to
+  # match Chaquopy's embedded runtime (app/build.gradle python { version }).
+  android_platform_tests:
+    name: Platform tests (kolibri-installer-android)
+    needs: [pre_job, stage1_required_checks]
+    if: ${{ needs.pre_job.outputs.should_skip != 'true' && needs.stage1_required_checks.result == 'success' }}
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7.0.0
+        with:
+          fetch-depth: 0
+      - name: Set up uv
+        uses: astral-sh/setup-uv@v8.2.0
+        with:
+          enable-cache: true
+          cache-python: true
+      - name: Install Python 3.10
+        run: uv python install 3.10
+      - name: Install dependencies
+        # --all-packages keeps root kolibri's runtime + test deps in the shared
+        # venv alongside the kolibri-installer-android member; the tests import
+        # kolibri.core.tasks and run real jobs through morango storage.
+        run: uv sync --group test --python 3.10 --all-packages
+      - name: Run tests
+        run: uv run python -O -m pytest platforms/android/app/src/main/python/tests -c pytest.ini --color=no
   # Single stable check for branch protection. Skipped matrix jobs don't
   # produce per-version check runs, so matrix-suffixed required checks hang
   # as "Expected" on PRs that skip the tests; require these jobs instead.
@@ -414,6 +443,7 @@ jobs:
       - sync_extras_plugin_tests
       - opensearch_plugin_tests
       - platform_tests
+      - android_platform_tests
     if: always()
     runs-on: ubuntu-latest
     steps:

platforms/android/app/build.gradle

@@ -248,6 +248,8 @@ android {
 		main {
 			python {
 				srcDir "src/main/python"
+				// Host-only pytest tests must not be bundled into the APK.
+				exclude "tests/**"
 			}
 		}
 	}

platforms/android/app/src/main/python/android_app_plugin/kolibri_plugin.py

@@ -10,6 +10,7 @@
 from kolibri.core.device.hooks import GetOSUserHook
 from kolibri.core.tasks.hooks import JobHook
 from kolibri.core.tasks.job import Priority
+from kolibri.core.tasks.job import State
 from kolibri.plugins import KolibriPluginBase
 from kolibri.plugins.hooks import register_hook
 
@@ -55,6 +56,9 @@ def schedule(
         job,
         orm_job,
     ):
+        self._enqueue_task(job, orm_job)
+
+    def _enqueue_task(self, job, orm_job):
         if orm_job.id:
             delay = 0
             if orm_job.scheduled_time:
@@ -85,6 +89,9 @@ def schedule(
             job.update_worker_info(extra=request_id)
 
     def update(self, job, orm_job, state=None, **kwargs):
+        if state == State.QUEUED:
+            self._enqueue_task(job, orm_job)
+
         currentLocale = Locale.getDefault().toLanguageTag()
 
         status = job.status(currentLocale)

platforms/android/app/src/main/python/task_identity.py

@@ -0,0 +1,13 @@
+"""Canonical supervisor-id form for the Android task layer."""
+
+import uuid
+
+
+def supervisor_id_from_request(request_id):
+    """
+    WorkManager gives dashed UUIDs; morango stores supervisor_id as dashless
+    hex. Both the claim (taskworker) and the reconcile live set
+    (task_reconciler) must normalize here, or a completion write looks disowned
+    and strands the job in RUNNING.
+    """
+    return uuid.UUID(str(request_id)).hex

platforms/android/app/src/main/python/task_reconciler.py

@@ -7,6 +7,7 @@
 import os
 
 from java import jclass
+from task_identity import supervisor_id_from_request
 
 from kolibri.core.tasks.job import State
 from kolibri.core.tasks.main import job_storage
@@ -17,12 +18,29 @@
 Task = jclass("org.learningequality.Kolibri.task.Task")
 
 
-def _get_workmanager_job_ids():
+def _extract_work_ids(work_infos):
     """
-    Get all active job IDs from WorkManager (ENQUEUED and RUNNING states)
+    Extract Kolibri job IDs and WorkManager request IDs from WorkInfo objects.
 
-    Returns:
-        set: Set of job ID strings
+    Returns (job_ids, request_ids) as sets of strings.
+    """
+    job_ids = set()
+    request_ids = set()
+    for work_info in work_infos:
+        request_ids.add(supervisor_id_from_request(work_info.getId()))
+        tags = work_info.getTags()
+        # Tags mix worker-class FQNs with our own; an explicit prefix identifies
+        # ours rather than excluding known patterns.
+        for tag in tags.toArray():
+            if tag.startswith("kolibri:job:"):
+                job_ids.add(tag[len("kolibri:job:") :])
+
+    return job_ids, request_ids
+
+
+def _get_active_work():
+    """
+    Active WorkManager job IDs and request IDs (ENQUEUED + RUNNING).
     """
     WorkManager = jclass("androidx.work.WorkManager")
     WorkInfo = jclass("androidx.work.WorkInfo")
@@ -38,17 +56,7 @@ def _get_workmanager_job_ids():
     work_query = WorkQuery.fromStates(states)
     work_info_list = work_manager.getWorkInfos(work_query).get()
 
-    # Extract Kolibri job IDs from tags
-    # Tags include both worker class FQNs and our job ID tag set via Task.enqueueOnce
-    # We use an explicit prefix to identify our tags rather than excluding known patterns
-    job_ids = set()
-    for work_info in work_info_list.toArray():
-        tags = work_info.getTags()
-        for tag in tags.toArray():
-            if tag.startswith("kolibri:job:"):
-                job_ids.add(tag[len("kolibri:job:") :])
-
-    return job_ids
+    return _extract_work_ids(work_info_list.toArray())
 
 
 def _get_kolibri_active_jobs():
@@ -126,13 +134,29 @@ def _do_reconciliation():
     logger.info("Starting task reconciliation")
 
     try:
+        # First, clear ownership of any supervised jobs whose WorkManager
+        # request is no longer live, so they can be re-dispatched. The live
+        # set is the pre-reconcile snapshot of active WorkManager request ids.
+        _, request_ids = _get_active_work()
+        job_storage.reconcile_stalled_jobs(live_supervisor_ids=request_ids)
+        logger.info(
+            f"Reconciled stalled jobs against {len(request_ids)} live "
+            "supervisor ids"
+        )
+
         kolibri_jobs = _get_kolibri_active_jobs()
         kolibri_job_ids = set(kolibri_jobs.keys())
         logger.info(
             f"Found {len(kolibri_job_ids)} active jobs in Kolibri database"
         )
 
-        workmanager_job_ids = _get_workmanager_job_ids()
+        # Post-reconcile WorkManager snapshot: reconcile's requeue re-enqueues
+        # jobs via the QUEUED hook, so they should be visible here to spare the
+        # membership sync a redundant enqueue. Correctness does not hinge on the
+        # timing, though: Task.enqueueOnce uses enqueueUniqueWork(job_id,
+        # REPLACE), so a membership-sync re-enqueue is idempotent and collapses
+        # any duplicate to a single worker regardless of snapshot timing.
+        workmanager_job_ids, _ = _get_active_work()
         logger.info(
             f"Found {len(workmanager_job_ids)} active tasks in WorkManager"
         )

platforms/android/app/src/main/python/taskworker.py

@@ -2,6 +2,8 @@
 import os
 import threading
 
+from task_identity import supervisor_id_from_request
+
 from kolibri.core.tasks.worker import execute_job as kolibri_execute_job
 
 logger = logging.getLogger(__name__)
@@ -31,6 +33,7 @@ def execute_job(job_id, request_id):
             worker_process=str(os.getpid()),
             worker_thread=str(threading.get_ident()),
             worker_extra=str(request_id),
+            supervisor_id=supervisor_id_from_request(request_id),
         )
         logger.info(
             "Completed Kolibri task worker for job {} (request {})".format(

platforms/android/app/src/main/python/tests/__init__.py

@@ -0,0 +1,27 @@
+import os
+import sys
+import types
+from unittest.mock import MagicMock
+
+# Ensure the python source dir is importable so `import taskworker`,
+# `import task_reconciler`, and `import android_app_plugin.kolibri_plugin`
+# resolve when tests run from the repo root.
+_SOURCE_DIR = os.path.dirname(os.path.dirname(__file__))
+if _SOURCE_DIR not in sys.path:
+    sys.path.insert(0, _SOURCE_DIR)
+
+
+def _install_fake_module(name, **attrs):
+    if name in sys.modules:
+        return
+    module = types.ModuleType(name)
+    for attr, value in attrs.items():
+        setattr(module, attr, value)
+    sys.modules[name] = module
+
+
+# The Chaquopy Java bridge is the single hard boundary (the Android/Java
+# runtime). It does not exist off-device, so stub it. Everything else,
+# including first-party modules like auth, imports for real against these.
+_install_fake_module("java", jclass=MagicMock(return_value=MagicMock()))
+_install_fake_module("java.util", Locale=MagicMock())

platforms/android/app/src/main/python/tests/conftest.py

@@ -0,0 +1,46 @@
+from unittest import mock
+
+from android_app_plugin.kolibri_plugin import AndroidJobHook
+
+from kolibri.core.tasks.job import Priority
+from kolibri.core.tasks.job import State
+
+TASK_PATH = "android_app_plugin.kolibri_plugin.Task"
+
+
+def _make_orm_job():
+    orm_job = mock.MagicMock()
+    orm_job.id = "job-1"
+    orm_job.scheduled_time = None
+    orm_job.priority = Priority.REGULAR
+    return orm_job
+
+
+def _make_job():
+    job = mock.MagicMock()
+    job.func = "kolibri.x"
+    job.long_running = False
+    # No status → update()'s notification branch is skipped, isolating enqueue.
+    job.status.return_value = None
+    return job
+
+
+def test_update_on_queued_transition_reenqueues():
+    hook = AndroidJobHook()
+    with mock.patch(TASK_PATH) as Task:
+        hook.update(_make_job(), _make_orm_job(), state=State.QUEUED)
+    assert Task.enqueueOnce.called
+
+
+def test_update_on_non_queued_state_does_not_reenqueue():
+    hook = AndroidJobHook()
+    with mock.patch(TASK_PATH) as Task:
+        hook.update(_make_job(), _make_orm_job(), state=State.RUNNING)
+    assert not Task.enqueueOnce.called
+
+
+def test_schedule_still_reenqueues():
+    hook = AndroidJobHook()
+    with mock.patch(TASK_PATH) as Task:
+        hook.schedule(_make_job(), _make_orm_job())
+    assert Task.enqueueOnce.called

platforms/android/app/src/main/python/tests/test_kolibri_plugin.py

@@ -0,0 +1,51 @@
+"""
+End-to-end: a dashed request id must drive a job to COMPLETED against real
+storage, exercising the ownership fence the unit tests can't reach (see
+task_identity for why normalization is load-bearing).
+"""
+
+from django.test import TestCase
+from task_identity import supervisor_id_from_request
+
+from kolibri.core.tasks.job import Job
+from kolibri.core.tasks.job import State
+from kolibri.core.tasks.main import job_storage
+
+# Captured mid-execution to prove the job is owned while RUNNING, not just
+# that ownership is cleared afterwards.
+_observed = {}
+
+
+def _observe_ownership_task(**kwargs):
+    orm_job = job_storage.get_orm_job(_observed["job_id"])
+    _observed["state"] = orm_job.state
+    _observed["supervisor_id"] = orm_job.supervisor_id
+    return "ok"
+
+
+class TaskWorkerLifecycleTest(TestCase):
+    databases = "__all__"
+
+    def test_execute_via_request_id_reaches_completed(self):
+        import taskworker
+
+        job = Job(_observe_ownership_task)
+        job_id = job_storage.enqueue_job(job)
+        _observed.clear()
+        _observed["job_id"] = job_id
+
+        # A canonical (dashed) java.util.UUID.toString() request id.
+        request_id = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
+        taskworker.execute_job(job_id, request_id)
+
+        # Mid-execution: the job was RUNNING and owned by the normalized id.
+        self.assertEqual(_observed["state"], State.RUNNING)
+        self.assertEqual(
+            _observed["supervisor_id"],
+            supervisor_id_from_request(request_id),
+        )
+
+        orm_job = job_storage.get_orm_job(job_id)
+        self.assertEqual(orm_job.state, State.COMPLETED)
+        # Ownership is cleared on a terminal transition.
+        self.assertIsNone(orm_job.supervisor_id)