From 1c30b9bf59b8489f7e7c068e5859097e57bbf5c2 Mon Sep 17 00:00:00 2001 From: Vinay Bikkina <704511+vb3@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:20:19 -0700 Subject: [PATCH] fix(cron): scope in_channel thread_id clear to the live-send/seed gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in_channel flat-delivery clear was gated on `runtime_adapter is not None`, but the flat continuation session is seeded ONLY on the live-send path, which also requires a running event loop. When an adapter is present but the loop is absent/not-running, the live-send/seed block is skipped and delivery falls through to the standalone path — clearing thread_id there flattened an unseeded brief with no continuable session behind it (and bypassed the D6 capability check). Factor the live-send condition into `live_adapter_ready` and gate both the thread_id clear and the live-send block on it, so the clear can no longer drift from the seed. Add an adapter-present/no-running-loop regression test (verified it fails against the un-scoped clear). Addresses review r3609147550. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- cron/scheduler.py | 43 ++++++++++++++++--------- tests/cron/test_scheduler.py | 61 ++++++++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 14 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index b5b967b55994..9514e5efd784 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1579,6 +1579,19 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # Prefer the live adapter when the gateway is running — this supports E2EE # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. runtime_adapter = (adapters or {}).get(platform) + # The live-send path (which SEEDS the flat in_channel continuation + # session via _seed_cron_channel_session) needs not just a live adapter + # but a running event loop to schedule the async send onto. Compute that + # gate ONCE so the in_channel thread_id clear below stays in lockstep + # with the live-send/seed block further down (they used to drift): an + # adapter can be present while the loop is absent/not-running, in which + # case the live-send block is skipped and delivery falls through to the + # standalone path — which cannot seed the flat session (r3609147550). + live_adapter_ready = ( + runtime_adapter is not None + and loop is not None + and getattr(loop, "is_running", lambda: False)() + ) delivered = False target_errors = [] @@ -1612,7 +1625,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option ) in_channel_surface = False - if in_channel_surface and mirror_this_target and runtime_adapter is not None: + if in_channel_surface and mirror_this_target and live_adapter_ready: # Force flat delivery (D2): the continuable-channel target must # ignore any inherited origin/target thread_id, or the flat # continuable session seeded below (thread_id=None, via @@ -1621,18 +1634,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # reads `thread_id` and would otherwise route into the origin # thread instead of flat into the channel. # - # Scoped to EXACTLY the target that gets flat-seeded: the origin- - # continuable target (`mirror_this_target`) on a live, in_channel- - # capable adapter. The `runtime_adapter is not None` guard keeps - # this in lockstep with the D6 capability fail-safe (which only runs - # with a live adapter) and with the seed itself - # (`in_channel_surface and mirror_this_target`, inside the - # live-adapter block below): fan-out / broadcast / explicit-thread - # targets keep their thread_id (they are not continuable and are - # never seeded), and the standalone no-adapter path falls back to - # thread delivery — it can never seed the flat session, so - # flattening there would drop the brief out of any continuable lane - # while also bypassing the D6 capability check. Placed AFTER + # Gated on `live_adapter_ready` (adapter present AND a running loop) + # so the clear fires ONLY on the live-send path that actually seeds + # the flat session — the SAME condition as the live-send block + # below. `runtime_adapter is not None` alone is broader than that + # path: an adapter can be present while the event loop is absent or + # not running, in which case the live-send/seed block is skipped and + # delivery falls through to the standalone path. Clearing thread_id + # there would flatten a brief into a channel with NO seeded + # continuable session behind it (and bypass the D6 capability + # check), so the standalone fallback must keep the origin thread + # (review r3609147550). + # + # Fan-out / broadcast / explicit-thread targets keep their thread_id + # (they are not continuable and are never seeded). Placed AFTER # mirror_this_target / origin_user_id are computed above — those # need the ORIGINAL thread_id to match the origin conversation. thread_id = None @@ -1689,7 +1704,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option thread_id = new_thread_id opened_thread_id = new_thread_id - if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)(): + if live_adapter_ready: # Telegram topic routing (#22773, regression fixed #52060): a # ``telegram::`` cron target is # ambiguous — a forum-style topic in a private chat and a genuine diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 39064bdf837c..c51ca739b343 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -4826,6 +4826,67 @@ class TestCronContinuableSurfaceInChannel: "(no live adapter can seed a flat continuable session), not flatten" ) + def test_in_channel_adapter_present_but_loop_not_running_preserves_origin_thread(self): + """Regression (review r3609147550): the thread_id clear must be scoped to + the FULL live-send condition (adapter present AND a running loop), not + just ``runtime_adapter is not None``. An adapter can be present while the + event loop is absent/not-running — then the live-send block that seeds + the flat continuable session (``_seed_cron_channel_session``) is SKIPPED + and delivery falls through to the standalone path. Clearing thread_id in + that case would flatten an UNSEEDED brief (no continuable session behind + it) and bypass the D6 capability check, so the standalone fallback must + keep the origin thread. Bites an unscoped clear AND a partial fix that + adds ``loop is not None`` but omits ``loop.is_running()``.""" + from gateway.config import Platform + + adapter = self._slack_adapter(supports_inchannel=True) + pconfig = MagicMock() + pconfig.enabled = True + pconfig.extra = {"cron_continuable_surface": "in_channel"} + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.SLACK: pconfig} + + # Live adapter present, but the event loop is NOT running — the middle + # state between the live-send path and the no-adapter standalone path. + # ``runtime_adapter is not None`` is true here (so an unscoped clear + # would wrongly flatten); ``live_adapter_ready`` is false. + loop = MagicMock() + loop.is_running.return_value = False + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True})) as send_mock: + job = { + "id": "brief-job", + "name": "Daily Brief", + "deliver": "origin", + # attach_to_session=True → mirror_this_target is True, so the + # (buggy) clear guard would actually fire without the loop gate. + "attach_to_session": True, + "origin": { + "platform": "slack", "chat_id": "C123", + "user_id": "U_HUMAN", "thread_id": "999.888", + }, + } + _deliver_result( + job, "Here is today's brief.", + adapters={Platform.SLACK: adapter}, loop=loop, + ) + + # The live-send block never ran (loop not running): the flat session was + # never seeded and the adapter was never used to send. + adapter.send.assert_not_awaited() + adapter._session_store.get_or_create_session.assert_not_called() + # Standalone fallback must preserve the origin thread, not flatten. + send_mock.assert_called_once() + assert send_mock.call_args.kwargs.get("thread_id") == "999.888", ( + "in_channel delivery with an adapter but no running loop must fall " + "back to the origin thread — the live-send block that seeds the flat " + "continuable session never ran, so flattening would leave an " + "unseeded brief" + ) + def test_thread_mode_default_still_opens_thread(self): """G1 regression: the default (thread) mode is byte-identical — the thread-open branch still fires when no surface key is set."""