diff --git a/tests/tools/test_cronjob_run_background.py b/tests/tools/test_cronjob_run_background.py index a185cfc299b0c..d33858c8b6615 100644 --- a/tests/tools/test_cronjob_run_background.py +++ b/tests/tools/test_cronjob_run_background.py @@ -61,7 +61,7 @@ class TestBackgroundDispatch: run_started = threading.Event() run_release = threading.Event() - def slow_run_one_job(job): + def slow_run_one_job(job, **kw): run_started.set() assert run_release.wait(timeout=5.0) return True @@ -224,7 +224,7 @@ class TestInFlightDedupe: seen_during_run = {} - def probe_run(job): + def probe_run(job, **kw): seen_during_run["registered"] = "job-bg-09" in sched.get_running_job_ids() return True diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index 41fc0321e06a0..3ebd5c31480b6 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/tests/tools/test_cronjob_run_immediate.py @@ -11,8 +11,10 @@ into the calling agent's activity tracker — otherwise the gateway inactivity watchdog kills the parent turn at ~1800s. """ import json +import sys import threading import time +from types import SimpleNamespace from unittest.mock import patch from tools.cronjob_tools import cronjob, _execute_job_now @@ -49,6 +51,39 @@ class TestCronjobRunExecutesImmediately: assert res["success"] is False m_run.assert_not_called() + def test_execute_job_now_passes_live_gateway_context_to_delivery(self): + """Manual runs must deliver on the live gateway adapter's owning loop.""" + adapters = {"matrix": object()} + gateway_loop = object() + runner = SimpleNamespace(adapters=adapters, _gateway_loop=gateway_loop) + completed = {"id": "job-run-1", "last_status": "ok", "last_error": None} + + with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("gateway.run._gateway_runner_ref", return_value=runner), \ + patch("cron.scheduler.run_one_job", return_value=True) as m_run, \ + patch("tools.cronjob_tools.get_job", return_value=completed): + res = _execute_job_now(dict(_JOB)) + + assert res["success"] is True + m_run.assert_called_once_with( + _JOB, + adapters=adapters, + loop=gateway_loop, + ) + + def test_execute_job_now_remains_standalone_without_gateway(self): + """CLI-only runs retain the standalone delivery path.""" + completed = {"id": "job-run-1", "last_status": "ok", "last_error": None} + + with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch.dict(sys.modules, {"gateway.run": None}), \ + patch("cron.scheduler.run_one_job", return_value=True) as m_run, \ + patch("tools.cronjob_tools.get_job", return_value=completed): + res = _execute_job_now(dict(_JOB)) + + assert res["success"] is True + m_run.assert_called_once_with(_JOB, adapters=None, loop=None) + def test_execute_job_now_marks_failure_on_exception(self): """An exception during fire is captured, marked failed, not propagated.""" with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ @@ -74,7 +109,7 @@ class TestCronjobRunExecutesImmediately: set_activity_callback(record) try: - def slow_run(job): + def slow_run(job, **kw): # Deterministic: block until at least one heartbeat has fired # (bounded so a broken heartbeat can't hang the test). assert heartbeat_seen.wait(timeout=5.0), "no heartbeat within 5s" @@ -123,7 +158,7 @@ class TestCronjobRunExecutesImmediately: set_activity_callback(record) try: - def slow_run(job): + def slow_run(job, **kw): # Ceiling=0 → the very first wake stops the loop without # touching. Give it a couple of cycles to prove silence. time.sleep(0.2) @@ -156,7 +191,7 @@ class TestCronjobRunExecutesImmediately: set_activity_callback(flaky) try: - def slow_run(job): + def slow_run(job, **kw): # Block until a heartbeat AFTER the raising one has fired. assert second_beat.wait(timeout=5.0), \ "heartbeat stopped after one callback exception" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index cd2c0ffe43e02..96fc1da7d6f68 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -729,9 +729,23 @@ def _run_claimed_job(job: Dict[str, Any]) -> Dict[str, Any]: ) _heartbeat_thread.start() + # Manual runs invoked from a gateway agent execute outside the scheduler + # ticker, but they still share the process with the live platform + # adapters. Pass the gateway-owned adapter map and event loop through + # to run_one_job so delivery is scheduled on the loop that owns clients + # such as Matrix/aiohttp. Calling those clients from run_one_job's + # standalone asyncio.run() loop raises errors like "Timeout context + # manager should be used inside a task" and can break encrypted Matrix + # delivery (#61495 — salvaged from #63586 by @Fly-onlyone). + gateway_module = sys.modules.get("gateway.run") + runner_ref = getattr(gateway_module, "_gateway_runner_ref", None) + runner = runner_ref() if callable(runner_ref) else None + adapters = getattr(runner, "adapters", None) if runner is not None else None + gateway_loop = getattr(runner, "_gateway_loop", None) if runner is not None else None + try: try: - processed = run_one_job(job) + processed = run_one_job(job, adapters=adapters, loop=gateway_loop) finally: _heartbeat_stop.set() if _heartbeat_thread is not None: