diff --git a/cron/scheduler.py b/cron/scheduler.py index 60c7ef3d6ee7..5b76d688a277 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -205,7 +205,7 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None: "telegram", "discord", "slack", "whatsapp", "signal", "matrix", "mattermost", "homeassistant", "dingtalk", "feishu", "wecom", "wecom_callback", "weixin", "sms", "email", "webhook", "bluebubbles", - "qqbot", "yuanbao", + "qqbot", "yuanbao", "whatsapp_cloud", }) # Platforms that support a configured cron/notification home target, mapped to @@ -701,6 +701,102 @@ def _seed_cron_thread_session( ) +def _seed_cron_channel_session( + job: dict, + adapter, + platform_name: str, + chat_id: str, + mirror_text: str, + *, + is_dm: bool, + user_id: Optional[str], + chat_name: Optional[str] = None, +) -> bool: + """Seed the FLAT (thread_id=None) session for an ``in_channel`` cron delivery. + + The ``in_channel`` surface (D1/D2) delivers the brief flat into the channel + with no thread, so the continuation surface is the whole-channel / + whole-DM session keyed ``thread_id=None`` — the same bucket + ``reply_in_thread: false`` routes an inbound plain reply to. + + Unlike the thread path, the shipped delivery-mirror alone is NOT sufficient + here: ``mirror_to_session`` only APPENDS to a session that already EXISTS + (``_find_session_id`` → no-op when none matches), and a flat channel + ``(…, None)`` row is only created when a human posts a top-level message the + bot processes — a ``chat_postMessage`` cron delivery never goes through the + inbound handler, so the row is usually absent and the mirror silently drops + the brief (verified live: the brief never landed, the reply had no context). + So we CREATE the flat session row first, exactly like + ``_seed_cron_thread_session`` does for threads, then mirror into it. + + The session KEY must match what the user's later inbound reply resolves to + (``build_session_key``): + - **Channel** (``chat_type="group"``): key is + ``…:group::`` — user-isolated — so the seed MUST carry + the **origin's real ``user_id``** (the member who scheduled the job), NOT + a synthetic ``system:cron`` id, or the reply keys to a different session. + - **1:1 DM** (``chat_type="dm"``): the key is ``…:dm:`` and does + NOT embed ``user_id``, so any ``user_id`` resolves to the same session. + ``chat_type`` mirrors the inbound handler's own choice + (``"dm" if is_dm else "group"``, ``adapter.py``), so the seeded key is + byte-identical to the reply's key. + + Returns True if a seed row was created and the brief mirrored, else False + (caller falls back to the plain mirror). Best-effort — a delivery that + already succeeded is never failed by a seeding problem. + """ + text = (mirror_text or "").strip() + if not text: + return False + try: + from gateway.config import Platform + from gateway.session import SessionSource + + chat_type = "dm" if is_dm else "group" + session_store = getattr(adapter, "_session_store", None) + if session_store is not None: + try: + platform_enum = Platform(platform_name.lower()) + except (ValueError, KeyError): + platform_enum = None + if platform_enum is not None: + dest_source = SessionSource( + platform=platform_enum, + chat_id=str(chat_id), + chat_name=chat_name, + chat_type=chat_type, + user_id=str(user_id) if user_id else None, + thread_id=None, # flat — the whole-channel/DM session + ) + # Create the flat session row so the mirror has a target and the + # user's later plain reply joins the SAME session. + session_store.get_or_create_session(dest_source) + + from gateway.mirror import mirror_to_session + + ok = mirror_to_session( + platform_name, + str(chat_id), + f"[Cron delivery: {job.get('name') or job.get('id', 'cron')}]\n{text}", + source_label="cron", + thread_id=None, + user_id=str(user_id) if user_id else None, + role="user", + ) + if ok: + logger.info( + "Job '%s': seeded flat in_channel session on %s:%s (chat_type=%s)", + job.get("id", "?"), platform_name, chat_id, chat_type, + ) + return bool(ok) + except Exception as e: + logger.debug( + "Job '%s': seeding in_channel session failed for %s:%s: %s", + job.get("id", "?"), platform_name, chat_id, e, + ) + return False + + def _cron_job_origin_log_suffix(job: dict) -> str: """Return safe provenance details for security warnings about a cron job. @@ -1261,6 +1357,50 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option delivered = False target_errors = [] + # Continuable cron surface (D1/D2/D6): resolve the delivery surface for + # this platform generically from its config ``extra``. Default "thread" + # (today's behaviour, byte-identical). "in_channel" delivers the brief + # FLAT into the channel (no dedicated thread) so a plain channel reply + # continues the job in-context via the shared-channel session + # ``(platform, chat_id, None)`` — the same bucket ``reply_in_thread: + # false`` routes inbound channel messages to. The key is read + # generically here (any platform); the ``in_channel`` branch is gated on + # the adapter capability flag ``supports_inchannel_continuable`` so an + # unsupported platform fails SAFE to "thread" (Slack is the first + # consumer; "first consumer ≠ definition"). + surface_mode = "thread" + try: + surface_raw = (pconfig.extra or {}).get("cron_continuable_surface") + if surface_raw is not None and str(surface_raw).strip().lower() == "in_channel": + surface_mode = "in_channel" + except Exception: + surface_mode = "thread" + in_channel_surface = surface_mode == "in_channel" + if in_channel_surface and runtime_adapter is not None and not getattr( + runtime_adapter, "supports_inchannel_continuable", False + ): + # Fail safe (D6): platform has no in_channel continuation primitive. + logger.debug( + "Job '%s': cron_continuable_surface=in_channel not supported on " + "%s, using thread", + job.get("id", "?"), platform_name, + ) + in_channel_surface = False + + # For an in_channel delivery the flat continuation session is created + # explicitly below (the shipped mirror only APPENDS to an existing + # session, and the flat channel row is otherwise absent for a + # chat_postMessage delivery). ``is_dm`` selects the session chat_type so + # the seeded key matches the inbound reply's key: a 1:1 DM keys as + # ``dm`` (Slack DM channel ids start with "D"; or the origin says so), + # everything else as ``group`` (shared channel). ``inchannel_seeded`` + # suppresses the generic mirror below so the brief is not double-written. + origin_chat_type = str(origin.get("chat_type") or "").lower() + is_dm_target = origin_chat_type == "dm" or ( + not origin_chat_type and str(chat_id).startswith("D") + ) + inchannel_seeded = False + # Continuable cron (thread-preferred): when mirroring is enabled for the # origin target and the gateway is live, try to open a DEDICATED thread # for this job and deliver the brief into it. On thread-capable @@ -1269,10 +1409,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # continues with full context. On DM-only platforms (WhatsApp/Signal) # create_handoff_thread returns None and we fall back to mirroring into # the origin DM session (handled after delivery). Cf. _process_handoff. + # + # in_channel surface (D2): SKIP thread creation entirely — leave + # thread_id=None so the delivery posts flat, then + # ``_seed_cron_channel_session`` (below) CREATES the shared-channel + # session and mirrors the brief into it. The shipped mirror alone is + # NOT enough here: ``mirror_to_session`` only APPENDS to an existing + # session and a flat ``(platform, chat_id, None)`` row is otherwise + # absent for a ``chat_postMessage`` delivery, so the seed must create + # the row first (F5). thread_seeded = False opened_thread_id: Optional[str] = None if ( mirror_this_target + and not in_channel_surface and runtime_adapter is not None and loop is not None and not thread_id # never override an explicit origin thread/topic @@ -1510,10 +1660,21 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option chat_name=origin.get("chat_name"), ) thread_seeded = True + # in_channel surface: CREATE + seed the flat channel/DM + # session (the shipped mirror only appends to an existing + # session — the flat row is otherwise absent for a + # chat_postMessage delivery, so the brief would be lost). + if in_channel_surface and mirror_this_target and not thread_seeded: + inchannel_seeded = _seed_cron_channel_session( + job, runtime_adapter, platform_name, chat_id, + mirror_text, is_dm=is_dm_target, + user_id=origin_user_id, + chat_name=origin.get("chat_name"), + ) _maybe_mirror_cron_delivery( job, platform_name, chat_id, mirror_text, thread_id=thread_id, user_id=origin_user_id, - enabled=mirror_this_target and not thread_seeded, + enabled=mirror_this_target and not thread_seeded and not inchannel_seeded, ) except Exception as e: err_msg = f"live adapter delivery to {platform_name}:{chat_id} failed: {e}" @@ -1535,12 +1696,30 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # prevent "coroutine was never awaited" RuntimeWarning, then retry in a # fresh thread that has no running loop. coro.close() - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: - future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)) - result = future.result(timeout=30) + # The thread-pool fallback can itself raise (SMTP ConnectionError, + # future.result timeout, etc.). An exception raised inside this + # `except RuntimeError` block is NOT caught by the sibling + # `except Exception` below — it would escape _deliver_result() + # and crash the whole delivery loop, silently skipping every + # remaining target (#47163). Wrap the fallback in its own + # try/except so a per-target failure is logged and the loop + # continues to the next target. + try: + pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) + try: + future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)) + result = future.result(timeout=30) + finally: + pool.shutdown(wait=False) + except Exception as e: + msg = f"delivery to {platform_name}:{chat_id} failed: {e}" + logger.error("Job '%s': %s", job["id"], msg, exc_info=True) + target_errors.extend([msg]) + delivery_errors.extend(target_errors) + continue except Exception as e: msg = f"delivery to {platform_name}:{chat_id} failed: {e}" - logger.error("Job '%s': %s", job["id"], msg) + logger.error("Job '%s': %s", job["id"], msg, exc_info=True) target_errors.extend([msg]) delivery_errors.extend(target_errors) continue diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index 51e93c8a7a9f..cc8654333146 100644 --- a/gateway/platforms/webhook.py +++ b/gateway/platforms/webhook.py @@ -66,7 +66,7 @@ "telegram", "discord", "slack", "signal", "sms", "whatsapp", "matrix", "mattermost", "homeassistant", "email", "dingtalk", "feishu", "wecom", "wecom_callback", "weixin", "bluebubbles", - "qqbot", "yuanbao", + "qqbot", "yuanbao", "whatsapp_cloud", } DEFAULT_HOST = "0.0.0.0" diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 01bae7ed2d4b..7fcd01322f55 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -552,6 +552,33 @@ def test_all_token_case_insensitive(self, monkeypatch): assert platforms == ["discord", "telegram"], f"token={token!r} -> {platforms}" +class TestBuiltinDeliveryPlatforms: + """Built-in platforms must pass the ``_KNOWN_DELIVERY_PLATFORMS`` gate.""" + + def test_whatsapp_cloud_home_channel_resolves(self, monkeypatch): + """whatsapp_cloud is a built-in adapter (never in the plugin + registry), so the hardcoded set is its only admission path; without + it the home-channel target silently vanishes from the resolution.""" + from cron.scheduler import _resolve_delivery_targets + + monkeypatch.setenv("WHATSAPP_CLOUD_HOME_CHANNEL", "15551234567") + targets = _resolve_delivery_targets( + {"deliver": "whatsapp_cloud", "origin": None} + ) + assert targets == [ + {"platform": "whatsapp_cloud", "chat_id": "15551234567", "thread_id": None} + ] + + def test_every_home_target_platform_is_known(self): + """Anti-drift guard: every platform with a home-target env var must + be deliverable — a mismatch means cron delivery is silently dropped + (the resolver returns no target and records no delivery error).""" + from cron.scheduler import _HOME_TARGET_ENV_VARS, _is_known_delivery_platform + + for platform in _HOME_TARGET_ENV_VARS: + assert _is_known_delivery_platform(platform), platform + + class TestDeliverResultWrapping: """Verify that cron deliveries are wrapped with header/footer and no longer mirrored.""" @@ -4391,3 +4418,79 @@ def test_seed_channel_session_noop_on_empty_text(self): store.get_or_create_session.assert_not_called() mirror_mock.assert_not_called() + +class TestMultiTargetDeliveryContinuesOnFailure: + """When delivery to one target fails inside the standalone thread-pool + fallback, the loop must continue to the remaining targets (#47163). + + The fallback runs inside the `except RuntimeError` block of + `_deliver_result`. Before the fix, an exception raised there (SMTP + ConnectionError, future.result timeout) escaped the function entirely — + it is NOT caught by the sibling `except Exception` — crashing the loop + and silently dropping every subsequent target. + """ + + def _email_cfg(self): + from gateway.config import Platform + + pconfig = MagicMock() + pconfig.enabled = True + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.EMAIL: pconfig} + return mock_cfg + + def test_first_target_failure_does_not_crash_loop(self): + """First email target fails in the fallback; the second is still attempted.""" + job = { + "id": "multi-email-job", + "deliver": "email:a@example.com,email:b@example.com", + } + + with patch("gateway.config.load_gateway_config", return_value=self._email_cfg()), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run", side_effect=RuntimeError("no running loop")), \ + patch("concurrent.futures.ThreadPoolExecutor") as mock_pool_cls: + mock_pool = MagicMock() + mock_pool_cls.return_value = mock_pool + + fail_future = MagicMock() + fail_future.result.side_effect = ConnectionError("SMTP connection refused") + ok_future = MagicMock() + ok_future.result.return_value = {"success": True} + mock_pool.submit.side_effect = [fail_future, ok_future] + + result = _deliver_result(job, "Report content") + + # Both targets attempted — the loop did not crash after the first failure. + assert mock_pool.submit.call_count == 2, ( + f"expected 2 delivery attempts, got {mock_pool.submit.call_count}" + ) + # First target's failure is surfaced in the returned error string. + assert result is not None + assert "a@example.com" in result + assert "SMTP connection refused" in result + + def test_all_targets_fail_returns_combined_errors(self): + """When every target fails, the result reports all of them.""" + job = { + "id": "all-fail-job", + "deliver": "email:a@example.com,email:b@example.com", + } + + with patch("gateway.config.load_gateway_config", return_value=self._email_cfg()), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run", side_effect=RuntimeError("no running loop")), \ + patch("concurrent.futures.ThreadPoolExecutor") as mock_pool_cls: + mock_pool = MagicMock() + mock_pool_cls.return_value = mock_pool + + fail_future = MagicMock() + fail_future.result.side_effect = ConnectionError("connection refused") + mock_pool.submit.return_value = fail_future + + result = _deliver_result(job, "Report content") + + assert result is not None + assert "a@example.com" in result + assert "b@example.com" in result + assert mock_pool.submit.call_count == 2 diff --git a/tests/gateway/test_webhook_adapter.py b/tests/gateway/test_webhook_adapter.py index 9e5340042ee3..913abc3e04f3 100644 --- a/tests/gateway/test_webhook_adapter.py +++ b/tests/gateway/test_webhook_adapter.py @@ -1009,6 +1009,35 @@ async def test_no_thread_id_sends_no_metadata(self): ) +class TestBuiltinDeliverPlatforms: + """send()'s admission gate for built-in (non-plugin) platforms.""" + + @pytest.mark.asyncio + async def test_whatsapp_cloud_passes_the_builtin_gate(self): + """whatsapp_cloud is a built-in adapter (never in the plugin + registry), so ``_BUILTIN_DELIVER_PLATFORMS`` is its only admission + path through ``send()``; without it the identical route config + fails "Unknown deliver type" while ``deliver_only`` routes (which + skip the gate via ``_direct_deliver``) work.""" + adapter = _make_adapter() + mock_target = AsyncMock() + mock_target.send = AsyncMock(return_value=SendResult(success=True)) + mock_runner = MagicMock() + mock_runner.adapters = {Platform("whatsapp_cloud"): mock_target} + mock_runner.config.get_home_channel.return_value = None + adapter.gateway_runner = mock_runner + + adapter._delivery_info["webhook:r:1"] = { + "deliver": "whatsapp_cloud", + "deliver_extra": {"chat_id": "15551234567"}, + } + result = await adapter.send("webhook:r:1", "hello") + assert result.success is True + mock_target.send.assert_awaited_once_with( + "15551234567", "hello", metadata=None + ) + + class TestInsecureNoAuthSafetyRail: """connect() refuses to start when INSECURE_NO_AUTH is combined with a non-loopback bind. Guards against accidentally exposing an unauthenticated