fix(cron): deliver to relay-fronted platforms via canonical home_channel (#84300)
Cron jobs targeting a relay-fronted logical platform (e.g. Discord behind
the relay connector) failed twice over:
1. Target resolution read only the legacy <PLATFORM>_HOME_CHANNEL env
mirror. The canonical home_channel block that /sethome persists to
config.yaml — the only store that exists in a relay-fronted deployment,
where no native env var is exported — was never consulted, so
deliver='discord' silently resolved to nothing and the job fell back
to local-only.
2. Even with a resolved target, the delivery loop's native
configured/enabled gate rejected the platform ('not configured/enabled')
although resolve_delivery_transport had already produced a live relay
transport fronting it. A relay-fronted platform is deliberately NOT
natively enabled (its credential lives in the connector), so the native
gate must not apply to a relay transport.
Resolution now falls back from the env mirror to
config.get_home_channel(platform) for both chat_id and thread_id (thread
affinity only when the chat id came from the same config block), which
also makes the 'all' routing token pick up relay-fronted platforms. The
delivery gate honours a resolved relay transport, mirroring the
enablement rule resolve_delivery_transport already applied; the standalone
(no-relay) path keeps the historical gate byte-identical.
This commit is contained in:
Ben Barclay
GitHub
parent
356c702b55
commit
76d832d385
|
|
@ -1178,8 +1178,35 @@ def _resolve_home_env_var(platform_name: str) -> str:
|
|||
return _plugin_cron_env_var(name)
|
||||
|
||||
|
||||
def _get_home_target_chat_id(platform_name: str) -> str:
|
||||
"""Return the configured home target chat/room ID for a delivery platform."""
|
||||
def _get_config_home_channel(platform_name: str):
|
||||
"""Return the persisted ``HomeChannel`` for a platform from gateway config.
|
||||
|
||||
``/sethome`` declares ``config.yaml`` canonical (it is the only store that
|
||||
survives for relay-fronted logical platforms, whose adapters are not
|
||||
natively enabled) and mirrors the value into the legacy
|
||||
``<PLATFORM>_HOME_CHANNEL`` env var only as a best-effort compatibility
|
||||
shim. Cron historically read ONLY the env mirror, so a home channel that
|
||||
existed solely in config.yaml — e.g. Discord fronted by the relay
|
||||
connector, where no ``DISCORD_HOME_CHANNEL`` was ever exported — was
|
||||
invisible and jobs silently fell back to local-only. Reading the
|
||||
canonical store here fixes that for every relay-fronted platform at once.
|
||||
"""
|
||||
try:
|
||||
from gateway.config import load_gateway_config, Platform
|
||||
|
||||
config = load_gateway_config()
|
||||
platform = Platform(platform_name.lower())
|
||||
return config.get_home_channel(platform)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"config home_channel lookup failed for platform %r",
|
||||
platform_name, exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _env_home_target_chat_id(platform_name: str) -> str:
|
||||
"""Return the home chat id from the legacy env mirror only (no config)."""
|
||||
env_var = _resolve_home_env_var(platform_name)
|
||||
if not env_var:
|
||||
return ""
|
||||
|
|
@ -1191,6 +1218,22 @@ def _get_home_target_chat_id(platform_name: str) -> str:
|
|||
return value
|
||||
|
||||
|
||||
def _get_home_target_chat_id(platform_name: str) -> str:
|
||||
"""Return the configured home target chat/room ID for a delivery platform.
|
||||
|
||||
Resolution order: platform env var (legacy mirror, kept first so an
|
||||
operator override keeps winning) → legacy env var name → the canonical
|
||||
``home_channel`` block persisted in config.yaml by ``/sethome``.
|
||||
"""
|
||||
value = _env_home_target_chat_id(platform_name)
|
||||
if value:
|
||||
return value
|
||||
home = _get_config_home_channel(platform_name)
|
||||
if home is not None and home.chat_id:
|
||||
return str(home.chat_id)
|
||||
return ""
|
||||
|
||||
|
||||
def _get_home_target_thread_id(platform_name: str) -> Optional[str]:
|
||||
"""Return the optional thread/topic ID for a platform home target.
|
||||
|
||||
|
|
@ -1203,18 +1246,26 @@ def _get_home_target_thread_id(platform_name: str) -> Optional[str]:
|
|||
without changing the lobby invariant.
|
||||
"""
|
||||
env_var = _resolve_home_env_var(platform_name)
|
||||
if not env_var:
|
||||
return None
|
||||
if platform_name.lower() == "telegram":
|
||||
cron_thread = os.getenv("TELEGRAM_CRON_THREAD_ID", "").strip()
|
||||
if cron_thread:
|
||||
return cron_thread
|
||||
value = os.getenv(f"{env_var}_THREAD_ID", "").strip()
|
||||
if not value:
|
||||
value = os.getenv(f"{env_var}_THREAD_ID", "").strip() if env_var else ""
|
||||
if not value and env_var:
|
||||
legacy = _LEGACY_HOME_TARGET_ENV_VARS.get(env_var)
|
||||
if legacy:
|
||||
value = os.getenv(f"{legacy}_THREAD_ID", "").strip()
|
||||
return value or None
|
||||
if value:
|
||||
return value
|
||||
# Canonical config.yaml fallback — same rationale as
|
||||
# _get_home_target_chat_id, and thread affinity only applies when the
|
||||
# chat itself resolved from the same config block (an env-provided chat
|
||||
# id keeps its env-provided thread semantics).
|
||||
if not _env_home_target_chat_id(platform_name):
|
||||
home = _get_config_home_channel(platform_name)
|
||||
if home is not None and home.thread_id:
|
||||
return str(home.thread_id)
|
||||
return None
|
||||
|
||||
|
||||
def _iter_home_target_platforms():
|
||||
|
|
@ -1741,7 +1792,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
|
|||
pconfig = config.platforms.get(platform)
|
||||
runtime_adapter = None
|
||||
|
||||
if not pconfig or not pconfig.enabled:
|
||||
if transport is not None and transport.is_relay:
|
||||
# A relay transport carries the RELAY adapter's config, and
|
||||
# resolve_delivery_transport already applied relay's enablement
|
||||
# rule (config block absent OR enabled). The logical platform is
|
||||
# deliberately NOT natively enabled in a relay-fronted deployment
|
||||
# (its credential lives in the connector), so the native
|
||||
# configured/enabled gate below must not apply — it used to
|
||||
# reject exactly the targets the relay was resolved to serve.
|
||||
if pconfig is None:
|
||||
from gateway.config import PlatformConfig
|
||||
pconfig = PlatformConfig(enabled=True)
|
||||
elif not pconfig or not pconfig.enabled:
|
||||
msg = f"platform '{platform_name}' not configured/enabled"
|
||||
logger.warning("Job '%s': %s", job["id"], msg)
|
||||
delivery_errors.append(msg)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,176 @@
|
|||
"""Cron delivery for relay-fronted logical platforms.
|
||||
|
||||
Bug report: a deployment where Discord is fronted by the relay connector
|
||||
(``GATEWAY_RELAY_PLATFORMS=discord``) could not use ``deliver='discord'``:
|
||||
|
||||
1. Resolution read ONLY the legacy ``DISCORD_HOME_CHANNEL`` env mirror, never
|
||||
the canonical ``platforms.discord.home_channel`` block that ``/sethome``
|
||||
persists to config.yaml — so the target silently resolved to nothing and
|
||||
the job fell back to local-only.
|
||||
2. Even with a resolved target, the delivery loop's native
|
||||
``pconfig.enabled`` gate rejected the platform ("not configured/enabled")
|
||||
although ``resolve_delivery_transport`` had already produced a live relay
|
||||
transport that fronts it — a relay-fronted logical platform is
|
||||
deliberately NOT natively enabled (its credential lives in the connector).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from concurrent.futures import Future
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from cron import scheduler as sched
|
||||
from cron.scheduler import (
|
||||
_deliver_result,
|
||||
_get_home_target_chat_id,
|
||||
_get_home_target_thread_id,
|
||||
_resolve_delivery_targets,
|
||||
)
|
||||
from gateway.config import HomeChannel, Platform
|
||||
|
||||
|
||||
def _gateway_config_with_home(platform=Platform.DISCORD, chat_id="1517373704248758474",
|
||||
thread_id=None):
|
||||
"""A gateway config whose ONLY home-channel source is config.yaml."""
|
||||
home = HomeChannel(platform=platform, chat_id=chat_id, name="Home",
|
||||
thread_id=thread_id)
|
||||
config = MagicMock()
|
||||
config.platforms = {}
|
||||
config.get_home_channel = lambda p: home if p == platform else None
|
||||
return config
|
||||
|
||||
|
||||
def _clear_home_env(monkeypatch):
|
||||
for var in ("DISCORD_HOME_CHANNEL", "DISCORD_HOME_CHANNEL_THREAD_ID"):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Resolution: config.yaml home_channel fallback
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestConfigHomeChannelFallback:
|
||||
def test_chat_id_falls_back_to_config_home_channel(self, monkeypatch):
|
||||
"""Env mirror empty → the canonical config.yaml home_channel is used."""
|
||||
_clear_home_env(monkeypatch)
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
return_value=_gateway_config_with_home()):
|
||||
assert _get_home_target_chat_id("discord") == "1517373704248758474"
|
||||
|
||||
def test_env_mirror_still_wins_over_config(self, monkeypatch):
|
||||
"""Operator env override keeps precedence over the config block."""
|
||||
monkeypatch.setenv("DISCORD_HOME_CHANNEL", "999")
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
return_value=_gateway_config_with_home()):
|
||||
assert _get_home_target_chat_id("discord") == "999"
|
||||
|
||||
def test_thread_id_falls_back_to_config_home_channel(self, monkeypatch):
|
||||
_clear_home_env(monkeypatch)
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
return_value=_gateway_config_with_home(thread_id="777")):
|
||||
assert _get_home_target_thread_id("discord") == "777"
|
||||
|
||||
def test_config_thread_not_used_when_chat_came_from_env(self, monkeypatch):
|
||||
"""Thread affinity: a config thread_id must not be grafted onto an
|
||||
env-provided chat id (they may point at different conversations)."""
|
||||
monkeypatch.setenv("DISCORD_HOME_CHANNEL", "999")
|
||||
monkeypatch.delenv("DISCORD_HOME_CHANNEL_THREAD_ID", raising=False)
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
return_value=_gateway_config_with_home(thread_id="777")):
|
||||
assert _get_home_target_thread_id("discord") is None
|
||||
|
||||
def test_no_source_returns_empty(self, monkeypatch):
|
||||
_clear_home_env(monkeypatch)
|
||||
config = MagicMock()
|
||||
config.platforms = {}
|
||||
config.get_home_channel = lambda p: None
|
||||
with patch("gateway.config.load_gateway_config", return_value=config):
|
||||
assert _get_home_target_chat_id("discord") == ""
|
||||
|
||||
def test_config_load_failure_fails_safe(self, monkeypatch):
|
||||
_clear_home_env(monkeypatch)
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
side_effect=RuntimeError("boom")):
|
||||
assert _get_home_target_chat_id("discord") == ""
|
||||
|
||||
def test_deliver_discord_resolves_via_config_home(self, monkeypatch):
|
||||
"""End to end: deliver='discord' on a job with no origin resolves a
|
||||
concrete target from the config.yaml home_channel alone."""
|
||||
_clear_home_env(monkeypatch)
|
||||
job = {"id": "j1", "deliver": "discord"}
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
return_value=_gateway_config_with_home()):
|
||||
targets = _resolve_delivery_targets(job)
|
||||
assert targets == [{
|
||||
"platform": "discord",
|
||||
"chat_id": "1517373704248758474",
|
||||
"thread_id": None,
|
||||
}]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Delivery: relay transport must bypass the native enabled gate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRelayDeliveryGate:
|
||||
def _relay_adapter(self):
|
||||
adapter = AsyncMock()
|
||||
adapter.fronts_platform = lambda p: p == Platform.DISCORD
|
||||
return adapter
|
||||
|
||||
def _job(self):
|
||||
return {
|
||||
"id": "relay-job",
|
||||
"name": "Relay Job",
|
||||
"deliver": "discord",
|
||||
"origin": {"platform": "discord", "chat_id": "123"},
|
||||
}
|
||||
|
||||
def _run(self, adapters, gateway_config):
|
||||
loop = MagicMock()
|
||||
loop.is_running.return_value = True
|
||||
|
||||
def fake_run_coro(coro, _loop):
|
||||
future = Future()
|
||||
try:
|
||||
future.set_result(asyncio.run(coro))
|
||||
except BaseException as e: # noqa: BLE001
|
||||
future.set_exception(e)
|
||||
return future
|
||||
|
||||
router = MagicMock()
|
||||
|
||||
async def _deliver_to_platform(target, content, metadata):
|
||||
return {"success": True, "raw_response": None}
|
||||
|
||||
router._deliver_to_platform = _deliver_to_platform
|
||||
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
return_value=gateway_config), \
|
||||
patch("cron.scheduler.load_config",
|
||||
return_value={"cron": {"wrap_response": False}}), \
|
||||
patch("gateway.delivery.DeliveryRouter", return_value=router), \
|
||||
patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
|
||||
return _deliver_result(self._job(), "Nightly report.",
|
||||
adapters=adapters, loop=loop)
|
||||
|
||||
def test_relay_fronted_platform_is_not_rejected(self, monkeypatch):
|
||||
"""A live relay transport that fronts Discord must deliver even though
|
||||
platforms.discord has no native config block at all."""
|
||||
_clear_home_env(monkeypatch)
|
||||
config = MagicMock()
|
||||
config.platforms = {} # neither discord nor relay configured natively
|
||||
config.get_home_channel = lambda p: None
|
||||
result = self._run({Platform.RELAY: self._relay_adapter()}, config)
|
||||
assert result is None # None == delivered without errors
|
||||
|
||||
def test_native_gate_preserved_without_relay(self, monkeypatch):
|
||||
"""No relay transport → the historical configured/enabled gate stays."""
|
||||
_clear_home_env(monkeypatch)
|
||||
config = MagicMock()
|
||||
config.platforms = {}
|
||||
config.get_home_channel = lambda p: None
|
||||
result = self._run({}, config)
|
||||
assert result is not None
|
||||
assert "not configured/enabled" in result
|
||||
Loading…
Reference in New Issue