fix(cron): deliver manual runs on gateway loop

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
 

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"

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: