SEP-1657: Record in-process dispatch runs and pin the recorder's contract (#1301)

Author: yyyyyyyan

app/tasks/celery.py

@@ -694,11 +694,14 @@ async def _dispatch_queue_item(
             await await_annotation(result, "STARTED")
         else:
             schedule_annotation(result, "STARTED")
-        return result
     finally:
         async with lock_session_maker() as async_session:
             await DispatchLockManager.delete(async_session, dispatch_lock)
 
+    if result.status.is_terminal():
+        await maybe_record_run(result.id, executor)
+    return result
+
 
 async def _raise_if_identical_task_conflict(
     queue_item: TaskHistory, session: AsyncSession

app/tasks/run_result.py

@@ -133,9 +133,19 @@ async def maybe_record_run(task_history_id: int, executor: BaseExecutor) -> None
     Best-effort: no-op when the history is missing, not terminal, or declares no
     recorder; and any failure (unresolvable path, unexpected executor error,
     recorder raising, DB error) is logged and swallowed so recording can never
-    fail the task-history sync.
-
-    :param task_history_id: The id of the just-synced ``TaskHistory``.
+    fail the caller.
+
+    Observation contract — coverage follows the entry point driving the
+    transition, not the status reached. Observed:
+    :func:`app.tasks.celery.sync_queue_item`, ``POST /history/{id}/sync/``, and
+    :func:`app.tasks.celery.dispatch_queue_item` for a backend whose dispatch
+    returns already-terminal (the Celery backend runs its callable inline). Not
+    observed, each pinned by a test: a run stopped via the stop route, one
+    failed before dispatch, and one the connectivity probe drives to terminal in
+    its own poll loop — the first two have no result to read by construction,
+    and the probe parses its own verdict.
+
+    :param task_history_id: The id of the just-terminal ``TaskHistory``.
     :param executor: The executor that ran the task, used to read its result.
     """
     recorder_path: str | None = None

tests/app/tasks/connectivity/test_service.py

@@ -24,6 +24,7 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.core.db.utils import get_async_session_maker_from_engine
+from app.tasks import hook_resolver
 from app.tasks.connectivity.constants import (
     PROVISIONING_TIMEOUT,
 )
@@ -205,6 +206,21 @@ async def run_python_task(self, session: AsyncSession) -> Task:
         )
         return await TaskManager.create(session, task_write)
 
+    @pytest_asyncio.fixture
+    async def recorder_run_python_task(self, session: AsyncSession) -> Task:
+        """Persist a ``run-python`` task row that declares a run-result recorder."""
+        task_write = TaskWrite.model_validate(
+            TaskFactory.build(
+                name="run-python",
+                backend=TaskBackendEnum.NOMAD,
+                is_template=False,
+                protected=False,
+                alert_on_fail=False,
+                run_result_recorder="pkg:rec",
+            )
+        )
+        return await TaskManager.create(session, task_write)
+
     @pytest_asyncio.fixture
     async def async_session_maker(self, session: AsyncSession):
         """Return a session-maker bound to the current test engine."""
@@ -311,6 +327,67 @@ async def sync_task_history(
             session, result.task_history_id
         )
 
+    async def test_terminal_run_records_nothing(
+        self,
+        session: AsyncSession,
+        recorder_run_python_task: Task,
+        async_session_maker,
+        mocker,
+    ) -> None:
+        """Skip recording for the probe's own run even when it reaches SUCCESS.
+
+        The poll loop drives ``sync_task_history`` directly rather than through
+        either sync seam, so a recorder declared on the shared ``run-python`` task
+        must not observe it — the probe parses its own verdict.
+        """
+        recorded = []
+
+        async def _recorder(db, history, result):
+            recorded.append(result)
+
+        request = _make_request(timeout=POLL_INTERVAL * 2)
+
+        async def sync_task_history(
+            queue_item: TaskHistory,
+            writer_session: AsyncSession | None = None,
+        ) -> TaskHistory:
+            assert writer_session is not None
+            await self._append_log(
+                writer_session,
+                queue_item.id,
+                TaskLogType.STDOUT,
+                json.dumps({"success": True}),
+            )
+            queue_item.status = TaskHistoryStatusEnum.SUCCESS
+            return queue_item
+
+        mock_executor = MagicMock(spec=BaseExecutor)
+        mock_executor.sync_task_history = sync_task_history
+
+        mocker.patch.dict(hook_resolver._RESOLVED, {"pkg:rec": _recorder}, clear=True)
+        with (
+            patch(
+                "app.tasks.connectivity.service.dispatch_queue_item",
+                side_effect=self._real_dispatch_running,
+            ),
+            patch(
+                "app.tasks.connectivity.service.get_executor_for_task",
+                return_value=mock_executor,
+            ),
+            patch(
+                "app.tasks.connectivity.service.get_async_session_maker",
+                return_value=async_session_maker,
+            ),
+            patch(
+                "app.tasks.run_result.get_async_session_maker",
+                return_value=async_session_maker,
+            ),
+        ):
+            result = await check_connectivity(session, request)
+
+        assert result.success is True
+        assert recorded == []
+
     async def test_unresolvable_payload_fails_terminally(
         self,
         session: AsyncSession,

tests/app/tasks/test_run_result.py

@@ -30,7 +30,7 @@
 from app.core.db.utils import get_async_session_maker_from_engine
 from app.core.utils import json_serializer, utc_now
 from app.tasks import hook_resolver
-from app.tasks.celery import sync_queue_item
+from app.tasks.celery import dispatch_queue_item, sync_queue_item
 from app.tasks.crud import TaskHistoryManager, TaskManager
 from app.tasks.execution.exceptions import TaskDataNotFoundInExecutorError
 from app.tasks.execution.executors.nomad import NomadExecutor
@@ -620,3 +620,178 @@ async def _stop(session, item):
         assert stopped.status == TaskHistoryStatusEnum.STOPPED
         assert recorded == []
         executor.stream_file.assert_not_called()
+
+
+def _dispatch_to(target: TaskHistoryStatusEnum, *, result: dict | None = None):
+    """Return a fake executor whose ``dispatch_task`` lands the run on ``target``."""
+
+    async def _fake_dispatch(session, item, task=None):
+        del task
+        item.started_at = utc_now()
+        item.status = target
+        if target.is_terminal():
+            item.finished_at = utc_now()
+        return await TaskHistoryManager.save(session, item)
+
+    executor = _fake_executor(
+        _raising(_response_error(404))
+        if result is None
+        else _yielding(_result_bytes(result))
+    )
+    executor.dispatch_task = AsyncMock(side_effect=_fake_dispatch)
+    return executor
+
+
+async def _run_dispatch(mocker, maker, history_id, executor):
+    """Drive ``dispatch_queue_item`` for ``history_id`` through ``executor``.
+
+    ``schedule_annotation`` is patched out because it spawns an unawaited
+    ``asyncio.create_task``, which would otherwise leak PMM work past the test.
+    """
+    mocker.patch("app.tasks.celery.get_async_session_maker", return_value=maker)
+    mocker.patch("app.tasks.run_result.get_async_session_maker", return_value=maker)
+    mocker.patch("app.tasks.celery.get_executor_for_task", return_value=executor)
+    mocker.patch("app.tasks.celery.schedule_annotation")
+    async with maker() as session:
+        queue_item = await TaskHistoryManager.get_or_404(
+            session,
+            select_related=(TaskHistory.task,),
+            query_options=[undefer(TaskHistory.execution_request)],
+            id=history_id,
+        )
+        return await dispatch_queue_item(queue_item, session)
+
+
+class TestDispatchSeam:
+    """Cover the recorder firing through the in-process dispatch seam.
+
+    A backend that runs its callable inline reaches a terminal status without an
+    intervening sync, so the sync seams never see it.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _clear_cache(self, mocker):
+        """Reset the resolver cache before each test."""
+        mocker.patch.dict(hook_resolver._RESOLVED, {}, clear=True)
+
+    @pytest.mark.asyncio
+    async def test_records_run_result_on_in_process_success(self, mocker):
+        """Fire the recorder with the run's result when dispatch lands on SUCCESS."""
+        recorded = []
+
+        async def _recorder(session, history, result):
+            recorded.append(result)
+
+        mocker.patch.dict(hook_resolver._RESOLVED, {"pkg:rec": _recorder})
+        executor = _dispatch_to(TaskHistoryStatusEnum.SUCCESS, result=_RESULT)
+        async with _recorder_db(
+            recorder="pkg:rec", status=TaskHistoryStatusEnum.PENDING
+        ) as (maker, history_id):
+            dispatched = await _run_dispatch(mocker, maker, history_id, executor)
+
+        assert dispatched.status == TaskHistoryStatusEnum.SUCCESS
+        assert recorded == [_RESULT]
+
+    @pytest.mark.asyncio
+    async def test_records_none_on_in_process_failure(self, mocker):
+        """Fire the recorder with ``None`` when dispatch lands on FAILED."""
+        recorded = []
+
+        async def _recorder(session, history, result):
+            recorded.append(result)
+
+        mocker.patch.dict(hook_resolver._RESOLVED, {"pkg:rec": _recorder})
+        executor = _dispatch_to(TaskHistoryStatusEnum.FAILED)
+        async with _recorder_db(
+            recorder="pkg:rec", status=TaskHistoryStatusEnum.PENDING
+        ) as (maker, history_id):
+            dispatched = await _run_dispatch(mocker, maker, history_id, executor)
+
+        assert dispatched.status == TaskHistoryStatusEnum.FAILED
+        assert recorded == [None]
+
+    @pytest.mark.asyncio
+    async def test_does_not_record_when_dispatch_leaves_the_run_running(self, mocker):
+        """Skip the seam entirely for a backend that dispatches asynchronously."""
+        recorded = []
+
+        async def _recorder(session, history, result):
+            recorded.append(result)
+
+        mocker.patch.dict(hook_resolver._RESOLVED, {"pkg:rec": _recorder})
+        executor = _dispatch_to(TaskHistoryStatusEnum.RUNNING, result=_RESULT)
+        async with _recorder_db(
+            recorder="pkg:rec", status=TaskHistoryStatusEnum.PENDING
+        ) as (maker, history_id):
+            dispatched = await _run_dispatch(mocker, maker, history_id, executor)
+
+        assert dispatched.status == TaskHistoryStatusEnum.RUNNING
+        assert recorded == []
+        executor.stream_file.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_recorder_failure_cannot_fail_dispatch(self, mocker):
+        """Swallow a raising recorder so it cannot fail the dispatch it observes."""
+
+        async def _recorder(session, history, result):
+            raise RuntimeError("recorder exploded")
+
+        mocker.patch.dict(hook_resolver._RESOLVED, {"pkg:rec": _recorder})
+        executor = _dispatch_to(TaskHistoryStatusEnum.SUCCESS, result=_RESULT)
+        async with _recorder_db(
+            recorder="pkg:rec", status=TaskHistoryStatusEnum.PENDING
+        ) as (maker, history_id):
+            dispatched = await _run_dispatch(  # must not raise
+                mocker, maker, history_id, executor
+            )
+
+        assert dispatched.status == TaskHistoryStatusEnum.SUCCESS
+
+
+class TestDispatchFailureCarveOut:
+    """Cover the deliberate exclusion of the pre-dispatch failure path."""
+
+    @pytest.fixture(autouse=True)
+    def _clear_cache(self, mocker):
+        """Reset the resolver cache before each test."""
+        mocker.patch.dict(hook_resolver._RESOLVED, {}, clear=True)
+
+    @pytest.mark.asyncio
+    async def test_failed_dispatch_records_nothing(self, mocker):
+        """Skip recording a run that failed before it ever held an allocation."""
+        recorded = []
+
+        async def _recorder(session, history, result):
+            recorded.append(result)
+
+        mocker.patch.dict(hook_resolver._RESOLVED, {"pkg:rec": _recorder})
+        executor = _dispatch_to(TaskHistoryStatusEnum.SUCCESS, result=_RESULT)
+        async with _recorder_db(
+            recorder="pkg:rec", status=TaskHistoryStatusEnum.PENDING
+        ) as (maker, history_id):
+            mocker.patch("app.tasks.celery.get_async_session_maker", return_value=maker)
+            mocker.patch(
+                "app.tasks.run_result.get_async_session_maker", return_value=maker
+            )
+            mocker.patch(
+                "app.tasks.celery.get_executor_for_task", return_value=executor
+            )
+            mocker.patch(
+                "app.tasks.celery.alert_service.trigger", new_callable=AsyncMock
+            )
+            async with maker() as session:
+                queue_item = await TaskHistoryManager.get_or_404(
+                    session,
+                    select_related=(TaskHistory.task,),
+                    query_options=[undefer(TaskHistory.execution_request)],
+                    id=history_id,
+                )
+                queue_item.execution_request.payload = "file:///sep/missing-payload.py"
+                queue_item = await TaskHistoryManager.save(
+                    session, queue_item, flag_modified_fields=["execution_request"]
+                )
+                failed = await dispatch_queue_item(queue_item, session)
+
+        assert failed.status == TaskHistoryStatusEnum.FAILED
+        assert recorded == []
+        executor.dispatch_task.assert_not_called()