diff --git a/cron/scheduler.py b/cron/scheduler.py index eb43196a7dd4..4a2b80b97903 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1204,6 +1204,36 @@ 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 + # 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 @@ -1212,10 +1242,18 @@ 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, and let the existing + # origin-mirror (below) seed the shared-channel session (F5: for a + # channel-origin job with thread_id=None, _target_matches_origin matches + # and _maybe_mirror_cron_delivery seeds (platform, chat_id, None)). No + # new seed call is needed. 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 diff --git a/gateway/config.py b/gateway/config.py index b8adb930dbe4..3ecef630f93d 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -1008,6 +1008,8 @@ def _merge_platform_map(source_platforms: Any) -> None: bridged["reply_prefix"] = platform_cfg["reply_prefix"] if "reply_in_thread" in platform_cfg: bridged["reply_in_thread"] = platform_cfg["reply_in_thread"] + if "cron_continuable_surface" in platform_cfg: + bridged["cron_continuable_surface"] = platform_cfg["cron_continuable_surface"] if "require_mention" in platform_cfg: bridged["require_mention"] = platform_cfg["require_mention"] if plat == Platform.TELEGRAM and "allowed_chats" in platform_cfg: diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 6e4db4467a09..15fc9eb3a764 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2257,6 +2257,21 @@ class BasePlatformAdapter(ABC): # "typed_command_prefix", "/"); no per-platform branching at call sites. typed_command_prefix: str = "/" + # Whether this adapter supports the ``in_channel`` continuable-cron surface + # (``platforms.

.extra.cron_continuable_surface: in_channel``): a + # continuable cron job delivered FLAT into a channel (no dedicated thread), + # with the user's plain channel reply continuing the job in-context via the + # shared-channel session. Only coherent on a platform that has BOTH a + # flat-reply outbound gate AND a whole-channel inbound session bucket keyed + # ``(platform, chat_id, None)`` — today that is Slack (``reply_in_thread: + # false``). Default False: an unsupported platform fails SAFE, treating + # ``in_channel`` as ``thread`` (a threaded continuation ≈ today's + # behaviour), never a dropped continuation. Read generically by the cron + # scheduler via ``getattr(adapter, "supports_inchannel_continuable", + # False)`` — no per-platform branching at the call site (the key stays a + # generic seam; Slack is merely the first consumer). + supports_inchannel_continuable: bool = False + def __init__(self, config: PlatformConfig, platform: Platform): self.config = config self.platform = platform diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index f2fdbd527b63..2767c6bc2fe3 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -422,6 +422,14 @@ class SlackAdapter(BasePlatformAdapter): # the prefix that works everywhere — instruction text must show it. typed_command_prefix = "!" + # Slack has both halves the ``in_channel`` continuable-cron surface needs: + # a flat-reply outbound gate (``reply_in_thread: false`` → ``_resolve_thread_ts`` + # returns None for top-level channel messages) AND a whole-channel inbound + # session bucket keyed ``(platform, channel_id, None)`` (the same + # ``reply_in_thread: false`` path in ``_handle_slack_message``). So a + # continuable cron delivered flat here continues in-context on a plain reply. + supports_inchannel_continuable = True + def __init__(self, config: PlatformConfig): super().__init__(config, Platform.SLACK) self._app: Optional[Any] = None @@ -1068,6 +1076,7 @@ async def connect(self, *, is_reconnect: bool = False) -> bool: self._warn_if_missing_group_dm_scopes(auth_response, team_name) self._warn_if_not_bot_token(auth_response, team_name) + self._warn_if_inchannel_without_flat_reply(team_name) # Register message event handler @self._app.event("message") @@ -1539,6 +1548,62 @@ def _dm_top_level_threads_as_sessions(self) -> bool: return True # default: each DM thread is its own session return str(raw).strip().lower() in {"1", "true", "yes", "on"} + def _cron_continuable_surface(self) -> str: + """Resolve the continuable-cron delivery surface for this platform. + + Values: ``"thread"`` (default — today's behaviour: a continuable cron + job opens a dedicated hidden thread and seeds it) or ``"in_channel"`` + (deliver FLAT into the channel timeline; the shared-channel session + ``(slack, channel_id, None)`` is the continuation surface). Set + ``platforms.slack.extra.cron_continuable_surface: in_channel`` in + config.yaml. Pair with ``reply_in_thread: false`` so the user's reply + is answered flat in the channel and keyed to the same shared session — + see ``_warn_if_inchannel_without_flat_reply``. Any unrecognised value + coerces to ``"thread"`` (fail safe). + """ + raw = self.config.extra.get("cron_continuable_surface") + if raw is None: + return "thread" + val = str(raw).strip().lower() + return "in_channel" if val == "in_channel" else "thread" + + def _warn_if_inchannel_without_flat_reply(self, team_name: str) -> None: + """Warn when ``in_channel`` is set without the required ``reply_in_thread: false`` pairing. + + The two knobs are orthogonal (D4/D5): ``cron_continuable_surface: + in_channel`` skips thread creation on delivery, and ``reply_in_thread: + false`` makes the bot answer inbound channel messages flat and key them + to the whole-channel session ``(slack, channel_id, None)``. For a + continuable in-channel cron to actually continue on a plain reply, BOTH + must hold: the seed lands in the shared-channel session, and the reply + must resolve to (and be answered in) that same flat session. + + Enforcement is WARN, not hard-require (D5): the misconfiguration fails + SAFE — ``in_channel`` without ``reply_in_thread: false`` yields a + threaded continuation (≈ today's behaviour), never a dropped/orphaned + session — so a config-load rejection would be heavier than warranted + and would make the two knobs non-orthogonal. Mirrors the existing + connect-time warning pattern (``_warn_if_missing_group_dm_scopes``, + ``_warn_if_not_bot_token``). + """ + try: + if self._cron_continuable_surface() != "in_channel": + return + # reply_in_thread defaults True (legacy: reply in a thread). + if self.config.extra.get("reply_in_thread", True): + logger.warning( + "[Slack] %s: cron_continuable_surface=in_channel is set " + "WITHOUT reply_in_thread=false. A continuable in-channel " + "cron job will deliver flat, but the bot will still reply " + "to your continuation in a thread — so it falls back to a " + "threaded continuation (\u2248 default behaviour), not the " + "flat channel session you asked for. Set " + "platforms.slack.extra.reply_in_thread: false to pair them.", + team_name, + ) + except Exception: + pass + def _resolve_thread_ts( self, reply_to: Optional[str] = None, diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 935533b11b97..73d172f23601 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -3997,3 +3997,134 @@ def test_seed_thread_session_noop_on_empty_text(self): ) store.get_or_create_session.assert_not_called() mirror_mock.assert_not_called() + + +class TestCronContinuableSurfaceInChannel: + """cron_continuable_surface: in_channel — deliver a continuable cron FLAT + into a channel (no dedicated thread), so a plain channel reply continues the + job via the shared-channel session (platform, chat_id, None). + + Design: decisions.md D1/D2/D6 + F5. The scheduler reads the per-platform key + generically from pconfig.extra; the in_channel branch is gated on the + adapter capability flag ``supports_inchannel_continuable`` (Slack=True, + others fail SAFE to thread). In in_channel mode the thread-open branch is + SKIPPED (thread_id stays None), so the existing origin-mirror seeds the + shared-channel session — no new seed code (G6). + """ + + def _slack_cfg(self, extra): + """A mock GatewayConfig with a Slack pconfig carrying ``extra``.""" + from gateway.config import Platform + + pconfig = MagicMock() + pconfig.enabled = True + pconfig.extra = extra + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.SLACK: pconfig} + return mock_cfg + + def _run_inchannel_delivery(self, extra, adapter, *, mirror_ok=True): + """Drive _deliver_result down the live-adapter path for a Slack + channel-origin job with the given ``extra`` config. Returns the + _open_continuable_cron_thread mock and the mirror_to_session mock.""" + from gateway.config import Platform + from concurrent.futures import Future + + mock_cfg = self._slack_cfg(extra) + + loop = MagicMock() + loop.is_running.return_value = True + + def fake_run_coro(coro, _loop): + future = Future() + try: + import asyncio as _asyncio + future.set_result(_asyncio.run(coro)) + except BaseException as _e: # noqa: BLE001 + future.set_exception(_e) + return future + + job = { + "id": "brief-job", + "name": "Daily Brief", + "deliver": "origin", + # Channel origin: no thread_id (flat channel message scheduled it). + "origin": {"platform": "slack", "chat_id": "C123"}, + # Opt into the continuable mirror. + "attach_to_session": True, + } + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("cron.scheduler._open_continuable_cron_thread") as open_thread_mock, \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), \ + patch("gateway.mirror.mirror_to_session", return_value=mirror_ok) as mirror_mock: + _deliver_result( + job, "Here is today's brief.", + adapters={Platform.SLACK: adapter}, loop=loop, + ) + return open_thread_mock, mirror_mock + + def _slack_adapter(self, supports_inchannel=True): + adapter = AsyncMock() + adapter.send.return_value = MagicMock(success=True) + # Capability flag read via getattr in the scheduler. + adapter.supports_inchannel_continuable = supports_inchannel + return adapter + + def test_in_channel_skips_thread_open(self): + """G2: in_channel mode must NOT open a handoff thread.""" + adapter = self._slack_adapter(supports_inchannel=True) + open_thread_mock, _ = self._run_inchannel_delivery( + {"cron_continuable_surface": "in_channel"}, adapter, + ) + open_thread_mock.assert_not_called() + + def test_in_channel_seeds_shared_channel_session_flat(self): + """G3/F5: with the thread-open branch skipped, the existing origin-mirror + seeds the shared-channel session with thread_id=None (flat).""" + adapter = self._slack_adapter(supports_inchannel=True) + _, mirror_mock = self._run_inchannel_delivery( + {"cron_continuable_surface": "in_channel"}, adapter, + ) + mirror_mock.assert_called_once() + # Seeded flat: no thread_id → session (slack, C123, None). + assert mirror_mock.call_args.kwargs.get("thread_id") is None + assert mirror_mock.call_args[0][0] == "slack" + assert mirror_mock.call_args[0][1] == "C123" + assert "Here is today's brief." in mirror_mock.call_args[0][2] + + 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.""" + adapter = self._slack_adapter(supports_inchannel=True) + open_thread_mock, _ = self._run_inchannel_delivery({}, adapter) + open_thread_mock.assert_called_once() + + def test_explicit_thread_value_opens_thread(self): + """An explicit cron_continuable_surface: thread is the default path.""" + adapter = self._slack_adapter(supports_inchannel=True) + open_thread_mock, _ = self._run_inchannel_delivery( + {"cron_continuable_surface": "thread"}, adapter, + ) + open_thread_mock.assert_called_once() + + def test_in_channel_on_unsupported_platform_fails_safe_to_thread(self): + """D6 fail-safe: in_channel on an adapter WITHOUT the capability flag + falls back to the thread path (a threaded continuation ≈ today), never + a dropped continuation.""" + adapter = self._slack_adapter(supports_inchannel=False) + open_thread_mock, _ = self._run_inchannel_delivery( + {"cron_continuable_surface": "in_channel"}, adapter, + ) + # Capability absent → treated as thread → thread-open still attempted. + open_thread_mock.assert_called_once() + + def test_unrecognised_surface_value_coerces_to_thread(self): + """Any non-'in_channel' value is the default thread path (fail safe).""" + adapter = self._slack_adapter(supports_inchannel=True) + open_thread_mock, _ = self._run_inchannel_delivery( + {"cron_continuable_surface": "bogus"}, adapter, + ) + open_thread_mock.assert_called_once() + diff --git a/tests/gateway/test_slack_cron_continuable_surface.py b/tests/gateway/test_slack_cron_continuable_surface.py new file mode 100644 index 000000000000..871df34ff6e1 --- /dev/null +++ b/tests/gateway/test_slack_cron_continuable_surface.py @@ -0,0 +1,149 @@ +""" +Tests for the Slack ``cron_continuable_surface`` extra key and its pairing warning. + +``cron_continuable_surface: in_channel`` (paired with ``reply_in_thread: false``) +lets a continuable cron job deliver FLAT into a channel — no dedicated thread — +so a plain channel reply continues the job via the shared-channel session +``(slack, channel_id, None)``. See specs/cron-inchannel-continuable decisions +D1/D4/D5/D6. + +- ``_cron_continuable_surface`` resolves the key: default ``"thread"``, coerces + any unrecognised value to ``"thread"`` (fail safe), only ``"in_channel"`` + opts in. +- ``supports_inchannel_continuable`` is True on Slack (it has both a flat-reply + outbound gate and a whole-channel inbound session bucket). +- ``_warn_if_inchannel_without_flat_reply`` warns (D5: warn, not hard-require) + when ``in_channel`` is set without ``reply_in_thread: false`` — the misconfig + fails SAFE to a threaded continuation, so it is a warning, not a rejection. +""" + +import logging +import sys +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Mock slack-bolt if not installed (same pattern as test_slack_user_token_warning.py) +# --------------------------------------------------------------------------- + +def _ensure_slack_mock(): + if "slack_bolt" in sys.modules and hasattr(sys.modules["slack_bolt"], "__file__"): + return + + slack_bolt = MagicMock() + slack_bolt.async_app.AsyncApp = MagicMock + slack_bolt.adapter.socket_mode.async_handler.AsyncSocketModeHandler = MagicMock + + slack_sdk = MagicMock() + slack_sdk.web.async_client.AsyncWebClient = MagicMock + + for name, mod in [ + ("slack_bolt", slack_bolt), + ("slack_bolt.async_app", slack_bolt.async_app), + ("slack_bolt.adapter", slack_bolt.adapter), + ("slack_bolt.adapter.socket_mode", slack_bolt.adapter.socket_mode), + ("slack_bolt.adapter.socket_mode.async_handler", + slack_bolt.adapter.socket_mode.async_handler), + ("slack_sdk", slack_sdk), + ("slack_sdk.web", slack_sdk.web), + ("slack_sdk.web.async_client", slack_sdk.web.async_client), + ]: + sys.modules.setdefault(name, mod) + + +_ensure_slack_mock() + +import plugins.platforms.slack.adapter as _slack_mod # noqa: E402 +_slack_mod.SLACK_AVAILABLE = True + +from plugins.platforms.slack.adapter import SlackAdapter # noqa: E402 + + +def _make_adapter(extra): + """object.__new__ skips __init__ (heavy setup) — established slack-test + pattern. Attach a minimal config carrying only the ``extra`` dict.""" + adapter = object.__new__(SlackAdapter) + cfg = MagicMock() + cfg.extra = dict(extra) + adapter.config = cfg + return adapter + + +# --- capability flag ------------------------------------------------------- + +def test_slack_declares_inchannel_capability(): + """Slack has both halves the in_channel surface needs, so the class-level + capability flag the cron scheduler reads generically must be True.""" + assert SlackAdapter.supports_inchannel_continuable is True + + +# --- surface resolver ------------------------------------------------------ + +def test_surface_defaults_to_thread(): + adapter = _make_adapter({}) + assert adapter._cron_continuable_surface() == "thread" + + +def test_surface_in_channel_opts_in(): + adapter = _make_adapter({"cron_continuable_surface": "in_channel"}) + assert adapter._cron_continuable_surface() == "in_channel" + + +def test_surface_in_channel_case_and_whitespace_insensitive(): + adapter = _make_adapter({"cron_continuable_surface": " In_Channel "}) + assert adapter._cron_continuable_surface() == "in_channel" + + +def test_surface_explicit_thread(): + adapter = _make_adapter({"cron_continuable_surface": "thread"}) + assert adapter._cron_continuable_surface() == "thread" + + +def test_surface_unrecognised_value_coerces_to_thread(): + """Fail safe: any value that isn't 'in_channel' resolves to 'thread'.""" + adapter = _make_adapter({"cron_continuable_surface": "bogus"}) + assert adapter._cron_continuable_surface() == "thread" + + +# --- pairing warning (D5: warn, not hard-require) -------------------------- + +def test_warns_when_in_channel_without_flat_reply(caplog): + """in_channel set, reply_in_thread left at its True default → warn.""" + adapter = _make_adapter({"cron_continuable_surface": "in_channel"}) + with caplog.at_level(logging.WARNING): + adapter._warn_if_inchannel_without_flat_reply("Acme") + matched = [r for r in caplog.records + if "cron_continuable_surface=in_channel" in r.message + and "reply_in_thread=false" in r.message] + assert matched + + +def test_warns_when_in_channel_with_reply_in_thread_true(caplog): + """Explicit reply_in_thread: true alongside in_channel → still warn.""" + adapter = _make_adapter( + {"cron_continuable_surface": "in_channel", "reply_in_thread": True} + ) + with caplog.at_level(logging.WARNING): + adapter._warn_if_inchannel_without_flat_reply("Acme") + assert any("cron_continuable_surface=in_channel" in r.message + for r in caplog.records) + + +def test_no_warning_when_properly_paired(caplog): + """in_channel + reply_in_thread: false is the correct pairing → silent.""" + adapter = _make_adapter( + {"cron_continuable_surface": "in_channel", "reply_in_thread": False} + ) + with caplog.at_level(logging.WARNING): + adapter._warn_if_inchannel_without_flat_reply("Acme") + assert not any("cron_continuable_surface=in_channel" in r.message + for r in caplog.records) + + +def test_no_warning_when_surface_is_thread(caplog): + """Default thread surface never warns about the pairing.""" + adapter = _make_adapter({"reply_in_thread": True}) + with caplog.at_level(logging.WARNING): + adapter._warn_if_inchannel_without_flat_reply("Acme") + assert not any("cron_continuable_surface=in_channel" in r.message + for r in caplog.records) diff --git a/tests/gateway/test_slack_mention.py b/tests/gateway/test_slack_mention.py index 62210a69b7a6..c23da97e6f6e 100644 --- a/tests/gateway/test_slack_mention.py +++ b/tests/gateway/test_slack_mention.py @@ -501,6 +501,56 @@ def test_config_bridges_slack_reply_in_thread(monkeypatch, tmp_path): ) == "171.000" +def test_config_bridges_slack_cron_continuable_surface_toplevel(monkeypatch, tmp_path): + """The cron_continuable_surface key bridges from a top-level ``slack:`` block + into slack.extra, mirroring reply_in_thread (specs D1/D6).""" + from gateway.config import load_gateway_config + + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + "slack:\n" + " cron_continuable_surface: in_channel\n" + " reply_in_thread: false\n", + encoding="utf-8", + ) + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setenv("SLACK_BOT_TOKEN", "xoxb-test") + + config = load_gateway_config() + + slack_config = config.platforms[Platform.SLACK] + assert slack_config.extra.get("cron_continuable_surface") == "in_channel" + # The adapter resolver reads the bridged key. + adapter = SlackAdapter(slack_config) + assert adapter._cron_continuable_surface() == "in_channel" + + +def test_config_bridges_slack_cron_continuable_surface_nested(monkeypatch, tmp_path): + """The key also bridges from the nested ``platforms.slack.extra`` path.""" + from gateway.config import load_gateway_config + + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + "platforms:\n" + " slack:\n" + " enabled: false\n" + " extra:\n" + " cron_continuable_surface: in_channel\n", + encoding="utf-8", + ) + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setenv("SLACK_BOT_TOKEN", "xoxb-test") + + config = load_gateway_config() + + slack_config = config.platforms[Platform.SLACK] + assert slack_config.extra.get("cron_continuable_surface") == "in_channel" + + def test_config_bridges_slack_strict_mention(monkeypatch, tmp_path): from gateway.config import load_gateway_config diff --git a/tests/manual/cron_inchannel_e2e.py b/tests/manual/cron_inchannel_e2e.py new file mode 100644 index 000000000000..aaa525f191c1 --- /dev/null +++ b/tests/manual/cron_inchannel_e2e.py @@ -0,0 +1,176 @@ +""" +Offline E2E harness for continuable in-channel cron (specs/cron-inchannel-continuable). + +Drives BOTH legs of the feature against the REAL code paths — no network, no +Slack contact, no Socket Mode — and asserts they converge on the same +shared-channel session key ``(slack, , None)``: + + LEG 1 (delivery): cron.scheduler._deliver_result(...) with a live Slack + adapter + cron_continuable_surface=in_channel → the thread-open branch is + SKIPPED and the shipped origin-mirror seeds (slack, C, None) with + thread_id=None (F5). Asserted via the mirror_to_session call. + + LEG 2 (reply): SlackAdapter._handle_slack_message(...) for a plain top-level + channel message under reply_in_thread=false → the inbound session keying + stamps thread_id=None, i.e. the SAME (slack, C, None) bucket the seed landed + in. Asserted via the dispatched MessageEvent.source.thread_id. + +If both legs report thread_id=None for the same channel, a plain channel reply +after an in_channel cron delivery resolves to the seeded session with the brief +in context (G3) — with NO visible thread (G2). + +Run from INSIDE the worktree so the worktree's code loads, not the editable +main-checkout install: + + cd + PYTHONPATH="$PWD" ../../.venv/bin/python tests/manual/cron_inchannel_e2e.py + +No real names anywhere (synthetic channel C_TEST / user U_TESTER / bot U_TESTBOT). +""" + +import asyncio +import sys +from concurrent.futures import Future +from unittest.mock import AsyncMock, MagicMock, patch + +# --- confirm we are running the WORKTREE's code, not the main checkout -------- +import cron.scheduler as _sched_mod +import plugins.platforms.slack.adapter as _slack_mod + +CHANNEL = "C_TEST" +BOT_UID = "U_TESTBOT" +USER_UID = "U_TESTER" +BRIEF = "Your daily brief: 3 PRs need review." + + +def leg1_delivery_seeds_flat_channel_session(): + """Real _deliver_result down the live-adapter path, in_channel mode.""" + from gateway.config import Platform + + # A Slack pconfig opting into in_channel. + pconfig = MagicMock() + pconfig.enabled = True + pconfig.extra = {"cron_continuable_surface": "in_channel", "reply_in_thread": False} + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.SLACK: pconfig} + + # A live Slack-like adapter that advertises the capability + sends OK. + adapter = AsyncMock() + adapter.send.return_value = MagicMock(success=True) + adapter.supports_inchannel_continuable = True + + loop = MagicMock() + loop.is_running.return_value = True + + def fake_run_coro(coro, _loop): + fut = Future() + try: + fut.set_result(asyncio.run(coro)) + except BaseException as e: # noqa: BLE001 + fut.set_exception(e) + return fut + + job = { + "id": "brief-job", + "name": "Daily Brief", + "deliver": "origin", + "origin": {"platform": "slack", "chat_id": CHANNEL}, # channel origin, no thread + "attach_to_session": True, + } + + open_thread_calls = [] + real_open = _sched_mod._open_continuable_cron_thread + + def _spy_open(*a, **k): + open_thread_calls.append((a, k)) + return real_open(*a, **k) + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("cron.scheduler._open_continuable_cron_thread", side_effect=_spy_open), \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + _sched_mod._deliver_result( + job, BRIEF, adapters={Platform.SLACK: adapter}, loop=loop, + ) + + assert not open_thread_calls, "LEG1 FAIL: thread-open was attempted in in_channel mode (G2)" + assert mirror_mock.call_count == 1, "LEG1 FAIL: brief was not mirrored/seeded" + kw = mirror_mock.call_args + seeded_platform = kw.args[0] + seeded_chat = kw.args[1] + seeded_text = kw.args[2] + seeded_thread = kw.kwargs.get("thread_id") + assert seeded_platform == "slack" and seeded_chat == CHANNEL, "LEG1 FAIL: wrong seed target" + assert seeded_thread is None, f"LEG1 FAIL: seed carried a thread_id ({seeded_thread!r}), not flat" + assert BRIEF in seeded_text, "LEG1 FAIL: brief text missing from seed" + return ("slack", seeded_chat, seeded_thread) + + +async def _leg2_reply_keys_flat_channel_session(): + """Real _handle_slack_message for a plain channel reply, reply_in_thread=false.""" + from gateway.config import PlatformConfig + + config = PlatformConfig(enabled=True, token="xoxb-test-not-a-real-token") + config.extra["reply_in_thread"] = False + # A channel where flat continuable-cron makes sense is one the bot answers + # ambiently — otherwise the user must @-mention on every reply (that is a + # pre-existing, orthogonal channel-gating choice, not part of this feature). + config.extra["require_mention"] = False + a = _slack_mod.SlackAdapter(config) + a._app = MagicMock() + a._app.client = AsyncMock() + a._bot_user_id = BOT_UID + a._running = True + + captured = [] + a.handle_message = AsyncMock(side_effect=lambda e: captured.append(e)) + + event = { + "channel": CHANNEL, + "channel_type": "channel", + "user": USER_UID, + # A plain channel reply — the user just types back, no @mention, no thread. + "text": "thanks, show me the first one", + "ts": "1700000000.000900", + } + + with patch.object(a, "_resolve_user_name", new=AsyncMock(return_value="tester")): + await a._handle_slack_message(event) + + assert len(captured) == 1, "LEG2 FAIL: plain channel reply was dropped (not continued)" + src = captured[0].source + assert src.thread_id is None, ( + f"LEG2 FAIL: reply keyed thread_id={src.thread_id!r}, not the flat " + "channel session — a threaded reply would NOT resolve to the seed" + ) + return ("slack", src.chat_id, src.thread_id) + + +def main(): + print(f"scheduler module: {_sched_mod.__file__}") + print(f"slack adapter module: {_slack_mod.__file__}") + if "cron-inchannel" not in _sched_mod.__file__: + print("WARNING: not running the worktree's scheduler — set PYTHONPATH=$PWD", file=sys.stderr) + + seed_key = leg1_delivery_seeds_flat_channel_session() + print(f"LEG 1 (delivery seed) → session key {seed_key}") + + reply_key = asyncio.run(_leg2_reply_keys_flat_channel_session()) + print(f"LEG 2 (inbound reply) → session key {reply_key}") + + # Convergence: both legs must land on (slack, CHANNEL, None). + assert seed_key[0] == reply_key[0], "platform mismatch" + assert str(seed_key[1]) == str(reply_key[1]), ( + f"channel mismatch: seed {seed_key[1]} vs reply {reply_key[1]}" + ) + assert seed_key[2] is None and reply_key[2] is None, "one leg was threaded" + print( + f"\nPASS: both legs converge on (slack, {CHANNEL}, None) — a plain " + "channel reply after an in_channel cron delivery continues the job " + "in-context, with no visible thread." + ) + + +if __name__ == "__main__": + main() diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md index cfbe48b40f05..b1f903d8d4f0 100644 --- a/website/docs/user-guide/features/cron.md +++ b/website/docs/user-guide/features/cron.md @@ -331,6 +331,45 @@ explicit other-chat deliveries) are never made continuable. The mirror is written as a labelled user turn (`[Cron delivery: ]`), which keeps the conversation history alternation-safe across all model providers. +#### Flat, in-channel continuation (Slack) + +The thread-preferred behaviour above mints a dedicated thread on every +delivery. If you'd rather have a continuable job land **flat in the channel +timeline** — no thread — set the Slack **continuable surface** to `in_channel`: + +```yaml +# ~/.hermes/config.yaml +slack: + cron_continuable_surface: in_channel # default: thread + reply_in_thread: false # required pairing (see below) + require_mention: false # so a plain reply continues the job +``` + +In `in_channel` mode the brief is delivered as an ordinary top-level channel +message (no thread is opened), and your reply continues the job via the +channel's shared session. Three settings work together: + +- **`cron_continuable_surface: in_channel`** — skips thread creation on delivery. +- **`reply_in_thread: false`** (required) — makes the bot answer your reply + *flat* in the channel and key it to the same whole-channel session the brief + was seeded into. Without it the continuation still works but arrives in a + thread (it falls back safely to thread-style continuation, never a dropped + reply — the gateway logs a warning at startup so you can spot the mismatch). +- **`require_mention: false`** (or add the channel to `free_response_channels`) + — so you can reply with a plain message; otherwise the bot only wakes when you + `@`-mention it on each reply. + +Because the continuation is the **whole-channel** session, it is shared: other +chatter in the channel — and a second continuable in-channel job — join the same +rolling conversation. That is inherent to "flat in a channel" and is the same +tradeoff `reply_in_thread: false` users already accept; use the default +`thread` surface when you want each delivery's follow-up isolated. + +This is a Slack capability today. Other platforms accept the key but fall back +to the `thread` surface (their continuation primitives differ); the choice is +per-platform, set under each platform's config. It's a gateway-side config flag +— a `/restart` picks it up; no Slack app reinstall is needed. + ### Silent suppression If the agent's final response contains `[SILENT]`, delivery is suppressed entirely. The output is still saved locally for audit (in `~/.hermes/cron/output/`), but no message is sent to the delivery target. diff --git a/website/docs/user-guide/messaging/slack.md b/website/docs/user-guide/messaging/slack.md index 67ccb6617339..474cb1600542 100644 --- a/website/docs/user-guide/messaging/slack.md +++ b/website/docs/user-guide/messaging/slack.md @@ -343,6 +343,13 @@ platforms: # (Slack's "Also send to channel" feature). # Only the first chunk of the first reply is broadcast. reply_broadcast: false + + # Continuable-cron delivery surface (default: "thread"). + # "in_channel" delivers a continuable cron job FLAT into the channel + # (no dedicated thread); pair with reply_in_thread: false (and + # require_mention: false) so a plain reply continues the job. + # See the cron guide → "Flat, in-channel continuation". + cron_continuable_surface: thread ``` | Key | Default | Description | @@ -350,6 +357,7 @@ platforms: | `platforms.slack.reply_to_mode` | `"first"` | Threading mode for multi-part messages: `"off"`, `"first"`, or `"all"` | | `platforms.slack.extra.reply_in_thread` | `true` | When `false`, channel messages get direct replies instead of threads. Messages inside existing threads still reply in-thread. | | `platforms.slack.extra.reply_broadcast` | `false` | When `true`, thread replies are also posted to the main channel. Only the first chunk is broadcast. | +| `platforms.slack.extra.cron_continuable_surface` | `"thread"` | Delivery surface for [continuable cron jobs](../features/cron.md#flat-in-channel-continuation-slack). `"thread"` opens a dedicated thread per delivery (default); `"in_channel"` delivers flat into the channel timeline. Pair `in_channel` with `reply_in_thread: false` (and `require_mention: false`) so a plain channel reply continues the job. | ### Session Isolation diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md index 985c28fb4742..1df1a55aa79d 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md @@ -319,6 +319,63 @@ cron: wrap_response: false ``` +### 可继续任务（回复 cron 投递） + +默认情况下，cron 投递是「发完即忘」的：消息发送出去，但不会进入聊天的对话历史， +因此如果你回复它，agent 并不记得自己说过什么。将任务设为**可继续**后，投递的简报 +就变成一段你可以回复进去的对话——agent 会把简报保留在上下文中，而不会反问 +「Task #2 是什么？」。 + +选择性启用，**默认关闭**。可在配置中全局启用，或通过 `cronjob` 工具的 +`attach_to_session` 按任务启用（会覆盖该任务的全局设置）： + +```yaml +# ~/.hermes/config.yaml +cron: + mirror_delivery: false # 设为 true 使 cron 投递可继续 +``` + +行为为**优先使用话题**，范围限定在任务的来源聊天： + +- **支持话题的平台**（Telegram 话题、Discord/Slack 话题）：每次投递都会新建 + 专用话题，并将简报植入该话题的会话中，因此在话题内回复即可带完整上下文继续。 +- **仅 DM 的平台**（WhatsApp、Signal、SMS）：不存在话题，因此简报会被镜像进 + 来源 DM 会话——DM 本身就是继续的载体。 + +只有来源聊天会被触及：扇出/广播目标（`all`、显式的其他聊天投递）永远不会被设为可继续。 + +#### 平铺频道内继续（Slack） + +上面的优先话题行为每次投递都会新建专用话题。如果你希望可继续任务**平铺落在频道 +时间线**中——不新建话题——将 Slack 的**继续投递方式**设为 `in_channel`： + +```yaml +# ~/.hermes/config.yaml +slack: + cron_continuable_surface: in_channel # 默认："thread" + reply_in_thread: false # 必需搭配（见下） + require_mention: false # 纯文本回复即可继续任务 +``` + +在 `in_channel` 模式下，简报作为普通的顶层频道消息投递（不新建话题），你的回复通过 +频道的共享会话继续任务。三项设置协同工作： + +- **`cron_continuable_surface: in_channel`**——投递时跳过新建话题。 +- **`reply_in_thread: false`**（必需）——让机器人在频道中*平铺*回复你，并将其 + 归入简报所植入的同一个整频道会话。缺少它时继续功能仍可用，但回复会出现在话题里 + （安全回退为话题式继续，绝不会丢失回复——网关会在启动时记录一条警告便于发现不匹配）。 +- **`require_mention: false`**（或将该频道加入 `free_response_channels`）——这样你 + 可以用纯文本消息回复；否则机器人只在你每次 `@` 提及它时才被唤醒。 + +由于继续载体是**整频道**会话，它是共享的：频道里的其他闲聊——以及第二个可继续的 +in_channel 任务——都会加入同一段滚动对话。这是「平铺在频道中」的固有取舍，与 +`reply_in_thread: false` 用户已经接受的取舍相同；若希望每次投递的后续讨论相互隔离， +请使用默认的 `thread` 方式。 + +这目前是 Slack 的能力。其他平台接受该键，但会回退到 `thread` 方式（它们的继续原语 +不同）；该选择按平台设置，位于各平台的配置下。这是网关侧的配置项——`/restart` 即可 +生效；无需重新安装 Slack 应用。 + ### 静默抑制 如果 agent 的最终响应以 `[SILENT]` 开头，投递将被完全抑制。输出仍会保存到本地以供审计（位于 `~/.hermes/cron/output/`），但不会向投递目标发送任何消息。 diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/slack.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/slack.md index 1fb03a1dc5f1..3df89baf14ae 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/slack.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/slack.md @@ -298,6 +298,13 @@ platforms: # （Slack 的"同时发送到频道"功能）。 # 仅广播第一条回复的第一个分块。 reply_broadcast: false + + # 可继续 cron 任务的投递方式（默认："thread"）。 + # "in_channel" 将可继续的 cron 任务直接平铺投递到频道中 + # （不新建话题）；需与 reply_in_thread: false（及 + # require_mention: false）搭配，纯文本回复即可继续任务。 + # 详见 cron 指南 →“平铺频道内继续”。 + cron_continuable_surface: thread ``` | 键 | 默认值 | 描述 | @@ -305,6 +312,7 @@ platforms: | `platforms.slack.reply_to_mode` | `"first"` | 多部分消息的话题模式：`"off"`、`"first"` 或 `"all"` | | `platforms.slack.extra.reply_in_thread` | `true` | 为 `false` 时，频道消息直接回复而非话题。已在话题中的消息仍在话题中回复。 | | `platforms.slack.extra.reply_broadcast` | `false` | 为 `true` 时，话题回复也会发布到主频道。仅广播第一个分块。 | +| `platforms.slack.extra.cron_continuable_surface` | `"thread"` | [可继续 cron 任务](../features/cron.md)的投递方式。`"thread"` 为每次投递新建专用话题（默认）；`"in_channel"` 直接平铺投递到频道时间线。使用 `in_channel` 时需搭配 `reply_in_thread: false`（及 `require_mention: false`），纯文本回复即可继续任务。 | ### 会话隔离