mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-14 06:13:15 +08:00
Compare commits
4 Commits
bb/desktop
...
fix/cron-i
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
91982408c3 | ||
|
|
26b776c046 | ||
|
|
14a3e280a1 | ||
|
|
6f499c729b |
@@ -645,6 +645,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:<chat_id>:<user_id>`` — 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:<chat_id>`` 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.
|
||||
|
||||
@@ -1205,6 +1301,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
|
||||
@@ -1213,10 +1353,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
|
||||
@@ -1454,10 +1602,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}"
|
||||
|
||||
@@ -1008,6 +1008,8 @@ def load_gateway_config() -> GatewayConfig:
|
||||
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:
|
||||
|
||||
@@ -2262,6 +2262,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.<p>.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
|
||||
|
||||
@@ -427,6 +427,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
|
||||
@@ -1073,6 +1081,7 @@ class SlackAdapter(BasePlatformAdapter):
|
||||
|
||||
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")
|
||||
@@ -1562,6 +1571,62 @@ class SlackAdapter(BasePlatformAdapter):
|
||||
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,
|
||||
|
||||
@@ -4142,3 +4142,250 @@ class TestCronDeliveryMirror:
|
||||
)
|
||||
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, origin=None):
|
||||
"""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).
|
||||
# Carries the scheduling user's id — the in_channel seed must key
|
||||
# the flat channel session to THIS user (see build_session_key).
|
||||
"origin": origin or {"platform": "slack", "chat_id": "C123", "user_id": "U_HUMAN"},
|
||||
# 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, with_store=True):
|
||||
adapter = AsyncMock()
|
||||
adapter.send.return_value = MagicMock(success=True)
|
||||
# Capability flag read via getattr in the scheduler.
|
||||
adapter.supports_inchannel_continuable = supports_inchannel
|
||||
# A live session store so the in_channel seed can CREATE the flat row
|
||||
# (the real bug: without a create step the mirror no-ops on a missing
|
||||
# session and the brief is lost). Use a plain MagicMock store.
|
||||
if with_store:
|
||||
adapter._session_store = MagicMock()
|
||||
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 (the real fix): in_channel CREATES the flat channel session row
|
||||
(thread_id=None) via the adapter's live store AND mirrors the brief into
|
||||
it. The prior implementation relied on the bare mirror, which no-ops
|
||||
when the flat row doesn't already exist — so the brief was silently lost
|
||||
(verified live). This asserts the create-then-mirror handoff."""
|
||||
adapter = self._slack_adapter(supports_inchannel=True)
|
||||
_, mirror_mock = self._run_inchannel_delivery(
|
||||
{"cron_continuable_surface": "in_channel"}, adapter,
|
||||
)
|
||||
# The flat session row must be CREATED (this is what was missing).
|
||||
adapter._session_store.get_or_create_session.assert_called_once()
|
||||
seeded = adapter._session_store.get_or_create_session.call_args[0][0]
|
||||
assert seeded.thread_id is None, "seed must be flat (thread_id=None)"
|
||||
assert seeded.chat_type == "group", "a channel (non-D) keys as group"
|
||||
assert str(seeded.chat_id) == "C123"
|
||||
assert str(seeded.user_id) == "U_HUMAN", (
|
||||
"channel session key embeds user_id — the seed MUST use the origin "
|
||||
"user's id or the inbound reply keys to a different session"
|
||||
)
|
||||
# Brief mirrored flat into that row.
|
||||
mirror_mock.assert_called_once()
|
||||
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_in_channel_dm_seeds_dm_session(self):
|
||||
"""1:1 DM (chat_id starts with 'D'): the flat session is created with
|
||||
chat_type='dm'. The DM session key does NOT embed user_id, so any
|
||||
user_id resolves to the same session — but chat_type must be 'dm' so the
|
||||
key prefix matches the inbound DM reply's key."""
|
||||
adapter = self._slack_adapter(supports_inchannel=True)
|
||||
_, mirror_mock = self._run_inchannel_delivery(
|
||||
{"cron_continuable_surface": "in_channel"}, adapter,
|
||||
origin={"platform": "slack", "chat_id": "D999", "user_id": "U_HUMAN"},
|
||||
)
|
||||
adapter._session_store.get_or_create_session.assert_called_once()
|
||||
seeded = adapter._session_store.get_or_create_session.call_args[0][0]
|
||||
assert seeded.chat_type == "dm", "a DM (chat_id starts with 'D') keys as dm"
|
||||
assert seeded.thread_id is None
|
||||
assert str(seeded.chat_id) == "D999"
|
||||
mirror_mock.assert_called_once()
|
||||
assert mirror_mock.call_args.kwargs.get("thread_id") is None
|
||||
|
||||
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()
|
||||
|
||||
# --- _seed_cron_channel_session: the create-then-mirror unit + the
|
||||
# KEY-MATCH invariant (seed key must equal the inbound reply's key) ---
|
||||
|
||||
def test_seed_channel_session_key_matches_inbound_channel_reply(self):
|
||||
"""The whole point: the flat session the seed CREATES must be keyed
|
||||
identically to what a plain inbound channel reply resolves to. Assert
|
||||
the invariant directly via build_session_key, not just call args."""
|
||||
from cron.scheduler import _seed_cron_channel_session
|
||||
from gateway.session import build_session_key, SessionSource
|
||||
from gateway.config import Platform
|
||||
|
||||
store = MagicMock()
|
||||
adapter = MagicMock()
|
||||
adapter._session_store = store
|
||||
|
||||
with patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock:
|
||||
ok = _seed_cron_channel_session(
|
||||
{"id": "j1", "name": "Brief"}, adapter, "slack", "C123",
|
||||
"Daily brief", is_dm=False, user_id="U_HUMAN", chat_name="ops",
|
||||
)
|
||||
assert ok is True
|
||||
seeded_source = store.get_or_create_session.call_args[0][0]
|
||||
seed_key = build_session_key(seeded_source)
|
||||
|
||||
# What a plain top-level channel reply (reply_in_thread:false → thread
|
||||
# None) from the same user resolves to:
|
||||
inbound = SessionSource(
|
||||
platform=Platform.SLACK, chat_id="C123", chat_type="group",
|
||||
user_id="U_HUMAN", thread_id=None,
|
||||
)
|
||||
assert seed_key == build_session_key(inbound), (
|
||||
f"seed key {seed_key} != inbound reply key {build_session_key(inbound)} "
|
||||
"— the reply would NOT continue the seeded session"
|
||||
)
|
||||
mirror_mock.assert_called_once()
|
||||
assert mirror_mock.call_args.kwargs.get("thread_id") is None
|
||||
assert mirror_mock.call_args.kwargs.get("user_id") == "U_HUMAN"
|
||||
|
||||
def test_seed_channel_session_key_matches_inbound_dm_reply(self):
|
||||
"""DM case: seeded key (chat_type=dm) equals the inbound DM reply key.
|
||||
The DM key ignores user_id, so a system id would also match — but
|
||||
chat_type MUST be 'dm' so the prefix aligns."""
|
||||
from cron.scheduler import _seed_cron_channel_session
|
||||
from gateway.session import build_session_key, SessionSource
|
||||
from gateway.config import Platform
|
||||
|
||||
store = MagicMock()
|
||||
adapter = MagicMock()
|
||||
adapter._session_store = store
|
||||
|
||||
with patch("gateway.mirror.mirror_to_session", return_value=True):
|
||||
_seed_cron_channel_session(
|
||||
{"id": "j1"}, adapter, "slack", "D999", "Daily brief",
|
||||
is_dm=True, user_id="U_HUMAN",
|
||||
)
|
||||
seeded_source = store.get_or_create_session.call_args[0][0]
|
||||
inbound = SessionSource(
|
||||
platform=Platform.SLACK, chat_id="D999", chat_type="dm",
|
||||
user_id="U_HUMAN", thread_id=None,
|
||||
)
|
||||
assert build_session_key(seeded_source) == build_session_key(inbound)
|
||||
assert seeded_source.chat_type == "dm"
|
||||
|
||||
def test_seed_channel_session_noop_on_empty_text(self):
|
||||
from cron.scheduler import _seed_cron_channel_session
|
||||
|
||||
store = MagicMock()
|
||||
adapter = MagicMock()
|
||||
adapter._session_store = store
|
||||
with patch("gateway.mirror.mirror_to_session") as mirror_mock:
|
||||
ok = _seed_cron_channel_session(
|
||||
{"id": "j1"}, adapter, "slack", "C123", " ",
|
||||
is_dm=False, user_id="U_HUMAN",
|
||||
)
|
||||
assert ok is False
|
||||
store.get_or_create_session.assert_not_called()
|
||||
mirror_mock.assert_not_called()
|
||||
|
||||
|
||||
149
tests/gateway/test_slack_cron_continuable_surface.py
Normal file
149
tests/gateway/test_slack_cron_continuable_surface.py
Normal file
@@ -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)
|
||||
@@ -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
|
||||
|
||||
|
||||
119
tests/manual/cron_inchannel_dm_e2e.py
Normal file
119
tests/manual/cron_inchannel_dm_e2e.py
Normal file
@@ -0,0 +1,119 @@
|
||||
"""
|
||||
DM-path verification for in_channel continuable cron (Option A scoping).
|
||||
|
||||
Option A: `cron_continuable_surface` is a CHANNEL feature. For a 1:1 DM the
|
||||
governing knob is the pre-existing `dm_top_level_threads_as_sessions` — a DM has
|
||||
no thread-vs-timeline split, so DM continuation works ONLY when top-level DMs
|
||||
share one flat session (`dm_top_level_threads_as_sessions: false`).
|
||||
|
||||
This harness PROVES that scoping against the REAL inbound handler
|
||||
(`SlackAdapter._handle_slack_message`) — no hard-coded thread_id assumption (the
|
||||
mistake that made the earlier E2E falsely pass):
|
||||
|
||||
SCENARIO 1 (the supported config, false): a top-level DM reply keys to the
|
||||
flat `…:dm:<chat>` session — the SAME key the cron seed
|
||||
(`_seed_cron_channel_session`, is_dm=True) creates. → continuation works.
|
||||
|
||||
SCENARIO 2 (the default, True — CONTROL): a top-level DM reply keys to a
|
||||
per-message `…:dm:<chat>:<ts>` session — DIVERGES from the flat seed. → this
|
||||
is exactly why in_channel does NOT give DM continuation under the default,
|
||||
and why Option A documents the requirement rather than pretending otherwise.
|
||||
|
||||
Run from INSIDE the worktree:
|
||||
cd <worktree>
|
||||
PYTHONPATH="$PWD" ../../.venv/bin/python tests/manual/cron_inchannel_dm_e2e.py
|
||||
|
||||
No real names. Uses a throwaway HERMES_HOME.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
os.environ["HERMES_HOME"] = tempfile.mkdtemp(prefix="cron_dm_e2e_")
|
||||
|
||||
import cron.scheduler as sched # noqa: E402
|
||||
from gateway.config import PlatformConfig, Platform # noqa: E402
|
||||
from gateway.session import build_session_key, SessionSource # noqa: E402
|
||||
from plugins.platforms.slack.adapter import SlackAdapter # noqa: E402
|
||||
|
||||
DM_CHAT = "D_TESTDM"
|
||||
BOT = "U_TESTBOT"
|
||||
USER = "U_TESTER"
|
||||
|
||||
|
||||
async def _inbound_dm_reply_key(dm_threads_as_sessions: bool):
|
||||
"""Drive the REAL _handle_slack_message for a top-level DM message and
|
||||
return (session_key, source) the dispatched MessageEvent resolves to."""
|
||||
cfg = PlatformConfig(enabled=True, token="xoxb-test-not-real")
|
||||
cfg.extra["dm_top_level_threads_as_sessions"] = dm_threads_as_sessions
|
||||
a = SlackAdapter(cfg)
|
||||
a._app = MagicMock()
|
||||
a._app.client = AsyncMock()
|
||||
a._bot_user_id = BOT
|
||||
a._running = True
|
||||
|
||||
captured = []
|
||||
a.handle_message = AsyncMock(side_effect=lambda e: captured.append(e))
|
||||
|
||||
event = {
|
||||
"channel": DM_CHAT,
|
||||
"channel_type": "im", # 1:1 DM
|
||||
"user": USER,
|
||||
"text": "how many items in that brief?",
|
||||
"ts": "1782999999.000100", # a NEW top-level DM message (no thread_ts)
|
||||
}
|
||||
with patch.object(a, "_resolve_user_name", new=AsyncMock(return_value="tester")):
|
||||
await a._handle_slack_message(event)
|
||||
|
||||
assert len(captured) == 1, "DM reply was dropped by the handler"
|
||||
src = captured[0].source
|
||||
return build_session_key(src), src
|
||||
|
||||
|
||||
def _seed_key() -> str:
|
||||
"""The session key the cron in_channel DM seed creates (is_dm=True, flat)."""
|
||||
seed_source = SessionSource(
|
||||
platform=Platform.SLACK, chat_id=DM_CHAT, chat_type="dm",
|
||||
user_id=USER, thread_id=None,
|
||||
)
|
||||
return build_session_key(seed_source)
|
||||
|
||||
|
||||
def main():
|
||||
print(f"adapter module: {SlackAdapter.__module__} ({sched.__file__.rsplit('/',2)[0]})")
|
||||
seed_key = _seed_key()
|
||||
print(f"\ncron in_channel DM seed key: {seed_key}")
|
||||
|
||||
# SCENARIO 1 — supported config (false): reply MUST converge on the seed.
|
||||
key_false, src_false = asyncio.run(_inbound_dm_reply_key(False))
|
||||
print(f"\n[dm_top_level_threads_as_sessions=false] reply key: {key_false}")
|
||||
print(f" thread_id on reply source: {src_false.thread_id!r}")
|
||||
assert key_false == seed_key, (
|
||||
f"FAIL: with the supported config, reply key {key_false} != seed {seed_key}"
|
||||
)
|
||||
print(" ✓ CONVERGES with the seed → DM continuation works")
|
||||
|
||||
# SCENARIO 2 — default (true): reply DIVERGES (this is why A documents the req).
|
||||
key_true, src_true = asyncio.run(_inbound_dm_reply_key(True))
|
||||
print(f"\n[dm_top_level_threads_as_sessions=true (default)] reply key: {key_true}")
|
||||
print(f" thread_id on reply source: {src_true.thread_id!r}")
|
||||
assert key_true != seed_key, (
|
||||
"unexpected: default DM keying matched the flat seed — the control is wrong"
|
||||
)
|
||||
print(" ✓ DIVERGES from the seed (per-message session) → in_channel gives")
|
||||
print(" NO DM continuation under the default; false is required (Option A)")
|
||||
|
||||
print(
|
||||
"\nPASS: Option A verified against the REAL inbound handler.\n"
|
||||
" • DM continuable cron works IFF dm_top_level_threads_as_sessions: false\n"
|
||||
" (reply and seed converge on the flat …:dm:<chat> session).\n"
|
||||
" • Under the default (true) they diverge — documented, not silently broken."
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
159
tests/manual/cron_inchannel_e2e.py
Normal file
159
tests/manual/cron_inchannel_e2e.py
Normal file
@@ -0,0 +1,159 @@
|
||||
"""
|
||||
Offline E2E for continuable in-channel cron (specs/cron-inchannel-continuable).
|
||||
|
||||
Exercises the REAL create→persist→find→append path end-to-end against a REAL
|
||||
SessionStore + REAL mirror_to_session + REAL _find_session_id + REAL
|
||||
build_session_key — NO mocking of the session layer. This is the harness that
|
||||
would have caught the shipped bug (the first version mocked mirror_to_session and
|
||||
so never exercised the fact that the mirror only APPENDS to a pre-existing
|
||||
session; the flat channel row was never created and the brief was silently lost).
|
||||
|
||||
Two scenarios, each asserting the brief actually lands in the SAME session the
|
||||
inbound reply resolves to:
|
||||
|
||||
CHANNEL: cron in_channel delivery → _seed_cron_channel_session CREATES the flat
|
||||
(slack, C, None) session (chat_type=group, keyed to the origin user) and
|
||||
mirrors the brief in. Then a plain channel reply (reply_in_thread:false →
|
||||
thread_id=None) keys to the SAME session → the brief is in its transcript.
|
||||
|
||||
1:1 DM: same, chat_type=dm. The DM session key ignores user_id, so the reply
|
||||
resolves regardless; assert the brief lands and the key matches.
|
||||
|
||||
Run from INSIDE the worktree (so the worktree code loads, not the editable
|
||||
main-checkout install):
|
||||
|
||||
cd <worktree>
|
||||
PYTHONPATH="$PWD" ../../.venv/bin/python tests/manual/cron_inchannel_e2e.py
|
||||
|
||||
Uses a throwaway HERMES_HOME so it never touches ~/.hermes. No real names.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _fresh_home():
|
||||
"""Point HERMES_HOME at a throwaway dir BEFORE importing gateway modules
|
||||
(mirror.py binds _SESSIONS_INDEX from get_hermes_home() at import time)."""
|
||||
d = tempfile.mkdtemp(prefix="cron_inchannel_e2e_")
|
||||
os.environ["HERMES_HOME"] = d
|
||||
return Path(d)
|
||||
|
||||
|
||||
HOME = _fresh_home()
|
||||
|
||||
# Import AFTER HERMES_HOME is set.
|
||||
import cron.scheduler as sched # noqa: E402
|
||||
import gateway.mirror as mirror # noqa: E402
|
||||
from gateway.config import GatewayConfig, Platform # noqa: E402
|
||||
from gateway.session import SessionStore, SessionSource, build_session_key # noqa: E402
|
||||
|
||||
# Force mirror.py's module-level index path to our temp home (it may have bound
|
||||
# a different get_hermes_home() at import if something imported it earlier).
|
||||
mirror._SESSIONS_DIR = HOME / "sessions"
|
||||
mirror._SESSIONS_INDEX = HOME / "sessions" / "sessions.json"
|
||||
|
||||
BRIEF = "brief: PRs need review\n- Harden: session lifecycle teardown"
|
||||
|
||||
|
||||
def _real_store():
|
||||
cfg = GatewayConfig()
|
||||
store = SessionStore(HOME / "sessions", cfg)
|
||||
return store
|
||||
|
||||
|
||||
def _run_scenario(name, chat_id, is_dm, reply_chat_type):
|
||||
print(f"\n=== {name} (chat_id={chat_id}, is_dm={is_dm}) ===")
|
||||
store = _real_store()
|
||||
|
||||
# A real Slack-like adapter exposing only what the seeder needs: the live
|
||||
# session store. (We call the seeder directly — the delivery leg's flat-post
|
||||
# is covered by the unit tests; here we prove the SESSION plumbing works.)
|
||||
class _Adapter:
|
||||
_session_store = store
|
||||
|
||||
ok = sched._seed_cron_channel_session(
|
||||
{"id": "brief-job", "name": "PR review brief"},
|
||||
_Adapter(), "slack", chat_id, BRIEF,
|
||||
is_dm=is_dm, user_id="U_HUMAN", chat_name="test",
|
||||
)
|
||||
assert ok, f"{name}: seeder returned False — session not created/mirrored"
|
||||
|
||||
# LEG 1: what session key did the seed create?
|
||||
seeded_source = SessionSource(
|
||||
platform=Platform.SLACK, chat_id=chat_id,
|
||||
chat_type="dm" if is_dm else "group",
|
||||
user_id="U_HUMAN", thread_id=None,
|
||||
)
|
||||
seed_key = build_session_key(seeded_source)
|
||||
|
||||
# LEG 2: what does a plain inbound reply (reply_in_thread:false → thread None)
|
||||
# from the same user resolve to?
|
||||
inbound = SessionSource(
|
||||
platform=Platform.SLACK, chat_id=chat_id, chat_type=reply_chat_type,
|
||||
user_id="U_HUMAN", thread_id=None,
|
||||
)
|
||||
reply_key = build_session_key(inbound)
|
||||
print(f" seed key : {seed_key}")
|
||||
print(f" reply key: {reply_key}")
|
||||
assert seed_key == reply_key, f"{name}: KEY MISMATCH — reply won't continue the seed"
|
||||
|
||||
# GROUND TRUTH: the brief must actually be in that session's transcript, and
|
||||
# discoverable via the same _find_session_id the inbound reply path uses.
|
||||
sid = mirror._find_session_id("slack", chat_id, thread_id=None, user_id="U_HUMAN")
|
||||
assert sid, f"{name}: _find_session_id found NO session — the reply would dead-end"
|
||||
# Read the session transcript back and confirm the brief text is present.
|
||||
idx = mirror._SESSIONS_INDEX
|
||||
import json
|
||||
data = json.loads(idx.read_text())
|
||||
entry = next((e for e in data.values() if isinstance(e, dict) and e.get("session_id") == sid), None)
|
||||
assert entry, f"{name}: session {sid} not in index"
|
||||
# transcript lives in the JSONL / SQLite; verify via the store's own read.
|
||||
found = _brief_in_transcript(store, sid)
|
||||
assert found, f"{name}: brief NOT found in session {sid} transcript"
|
||||
print(f" ✓ session {sid} created, brief present, reply resolves here")
|
||||
return True
|
||||
|
||||
|
||||
def _brief_in_transcript(store, sid):
|
||||
"""Best-effort read of the session transcript to confirm the brief landed."""
|
||||
# Try the SQLite DB first (the mirror writes both JSONL + SQLite).
|
||||
try:
|
||||
from hermes_state import SessionDB
|
||||
db = SessionDB()
|
||||
msgs = db.get_messages(sid)
|
||||
for m in msgs:
|
||||
if "PRs need review" in str(m.get("content", "")):
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
# Fallback: scan the JSONL transcript file.
|
||||
for p in (HOME / "sessions").glob("*.json*"):
|
||||
try:
|
||||
if "PRs need review" in p.read_text():
|
||||
return True
|
||||
except Exception:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
print(f"scheduler module: {sched.__file__}")
|
||||
print(f"HERMES_HOME (throwaway): {HOME}")
|
||||
if "cron-inchannel" not in sched.__file__:
|
||||
print("WARNING: not the worktree scheduler — set PYTHONPATH=$PWD", file=sys.stderr)
|
||||
|
||||
_run_scenario("CHANNEL", "C_TEST", is_dm=False, reply_chat_type="group")
|
||||
_run_scenario("1:1 DM", "D_TEST", is_dm=True, reply_chat_type="dm")
|
||||
|
||||
print(
|
||||
"\nPASS: in_channel cron seeds the flat session for BOTH a channel and a "
|
||||
"1:1 DM; the brief lands in the transcript and a plain reply resolves to "
|
||||
"the same session (continuation works)."
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -331,6 +331,63 @@ explicit other-chat deliveries) are never made continuable. The mirror is
|
||||
written as a labelled user turn (`[Cron delivery: <task name>]`), 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.
|
||||
|
||||
:::note 1:1 DMs
|
||||
`cron_continuable_surface` is a **channel** setting — a 1:1 DM has no
|
||||
thread-vs-timeline split to choose between (the DM is already flat), so the key
|
||||
has no effect there. What governs whether a DM cron delivery is continuable is
|
||||
the separate, pre-existing knob **`slack.dm_top_level_threads_as_sessions`**:
|
||||
|
||||
- **`false`** — all top-level DMs share one rolling DM session, so a continuable
|
||||
cron brief and your reply land in the **same** session and the job continues in
|
||||
context. This is what you want for continuable cron in a DM.
|
||||
- **`true`** (default) — each top-level DM message is its own session, so a reply
|
||||
to a delivered brief starts a *fresh* session that has no record of the brief.
|
||||
Continuation does not work in this mode (for cron or any other flat delivery).
|
||||
|
||||
So for a continuable cron job delivered to a 1:1 DM, set
|
||||
`slack.dm_top_level_threads_as_sessions: false`. `cron_continuable_surface` is
|
||||
not required (and is ignored) for DMs.
|
||||
:::
|
||||
|
||||
### 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.
|
||||
|
||||
@@ -344,6 +344,13 @@ platforms:
|
||||
# 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
|
||||
|
||||
# Render agent messages as Slack Block Kit blocks (default: false).
|
||||
# When true, the final agent message is sent with structured blocks —
|
||||
# section headers, dividers, true nested lists (via rich_text), and
|
||||
@@ -359,6 +366,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. |
|
||||
| `platforms.slack.extra.rich_blocks` | `false` | When `true`, agent messages are rendered as [Block Kit](https://docs.slack.dev/block-kit/) blocks (headers, dividers, true nested lists, and native tables). A plain-text fallback is always sent. Tables over Slack's limits fall back to aligned monospace. No app reinstall required — it's a send-side change only. |
|
||||
|
||||
### Session Isolation
|
||||
|
||||
@@ -319,6 +319,78 @@ 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 应用。
|
||||
|
||||
:::note 1:1 私信(DM)
|
||||
`cron_continuable_surface` 是**频道**设置——1:1 私信没有「话题 vs 时间线」的区分
|
||||
(私信本身就是平铺的),因此该键在私信中无效。决定私信 cron 投递是否可继续的是另一个
|
||||
已有的独立开关 **`slack.dm_top_level_threads_as_sessions`**:
|
||||
|
||||
- **`false`**——所有顶层私信共享同一个滚动私信会话,因此可继续的 cron 简报与你的回复
|
||||
落在**同一个**会话里,任务得以带上下文继续。这正是私信中可继续 cron 所需要的。
|
||||
- **`true`**(默认)——每条顶层私信消息各自成为独立会话,因此对已投递简报的回复会开启
|
||||
一个**全新**、不含该简报记录的会话。此模式下继续功能不可用(对 cron 或任何平铺投递皆然)。
|
||||
|
||||
所以,若要让 cron 任务在 1:1 私信中可继续,请设置
|
||||
`slack.dm_top_level_threads_as_sessions: false`。私信不需要(也会忽略)
|
||||
`cron_continuable_surface`。
|
||||
:::
|
||||
|
||||
### 静默抑制
|
||||
|
||||
如果 agent 的最终响应以 `[SILENT]` 开头,投递将被完全抑制。输出仍会保存到本地以供审计(位于 `~/.hermes/cron/output/`),但不会向投递目标发送任何消息。
|
||||
|
||||
@@ -299,6 +299,13 @@ platforms:
|
||||
# 仅广播第一条回复的第一个分块。
|
||||
reply_broadcast: false
|
||||
|
||||
# 可继续 cron 任务的投递方式(默认:"thread")。
|
||||
# "in_channel" 将可继续的 cron 任务直接平铺投递到频道中
|
||||
# (不新建话题);需与 reply_in_thread: false(及
|
||||
# require_mention: false)搭配,纯文本回复即可继续任务。
|
||||
# 详见 cron 指南 →“平铺频道内继续”。
|
||||
cron_continuable_surface: thread
|
||||
|
||||
# 将 Agent 消息渲染为 Slack Block Kit 区块(默认:false)。
|
||||
# 为 true 时,最终的 Agent 消息会以结构化区块发送——包括
|
||||
# 章节标题、分隔线、真正的嵌套列表(通过 rich_text)以及
|
||||
@@ -313,6 +320,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`),纯文本回复即可继续任务。 |
|
||||
| `platforms.slack.extra.rich_blocks` | `false` | 为 `true` 时,Agent 消息会渲染为 [Block Kit](https://docs.slack.dev/block-kit/) 区块(标题、分隔线、真正的嵌套列表以及原生表格)。始终附带纯文本回退。超出 Slack 限制的表格会回退为对齐的等宽文本。无需重新安装应用——这仅是发送端的改动。 |
|
||||
|
||||
### 会话隔离
|
||||
|
||||
Reference in New Issue
Block a user