diff --git a/cron/scheduler.py b/cron/scheduler.py index aa2749043f0..548f8342354 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -701,6 +701,102 @@ def _seed_cron_thread_session( ) +def _seed_cron_channel_session( + job: dict, + adapter, + platform_name: str, + chat_id: str, + mirror_text: str, + *, + is_dm: bool, + user_id: Optional[str], + chat_name: Optional[str] = None, +) -> bool: + """Seed the FLAT (thread_id=None) session for an ``in_channel`` cron delivery. + + The ``in_channel`` surface (D1/D2) delivers the brief flat into the channel + with no thread, so the continuation surface is the whole-channel / + whole-DM session keyed ``thread_id=None`` — the same bucket + ``reply_in_thread: false`` routes an inbound plain reply to. + + Unlike the thread path, the shipped delivery-mirror alone is NOT sufficient + here: ``mirror_to_session`` only APPENDS to a session that already EXISTS + (``_find_session_id`` → no-op when none matches), and a flat channel + ``(…, None)`` row is only created when a human posts a top-level message the + bot processes — a ``chat_postMessage`` cron delivery never goes through the + inbound handler, so the row is usually absent and the mirror silently drops + the brief (verified live: the brief never landed, the reply had no context). + So we CREATE the flat session row first, exactly like + ``_seed_cron_thread_session`` does for threads, then mirror into it. + + The session KEY must match what the user's later inbound reply resolves to + (``build_session_key``): + - **Channel** (``chat_type="group"``): key is + ``…:group::`` — user-isolated — so the seed MUST carry + the **origin's real ``user_id``** (the member who scheduled the job), NOT + a synthetic ``system:cron`` id, or the reply keys to a different session. + - **1:1 DM** (``chat_type="dm"``): the key is ``…:dm:`` and does + NOT embed ``user_id``, so any ``user_id`` resolves to the same session. + ``chat_type`` mirrors the inbound handler's own choice + (``"dm" if is_dm else "group"``, ``adapter.py``), so the seeded key is + byte-identical to the reply's key. + + Returns True if a seed row was created and the brief mirrored, else False + (caller falls back to the plain mirror). Best-effort — a delivery that + already succeeded is never failed by a seeding problem. + """ + text = (mirror_text or "").strip() + if not text: + return False + try: + from gateway.config import Platform + from gateway.session import SessionSource + + chat_type = "dm" if is_dm else "group" + session_store = getattr(adapter, "_session_store", None) + if session_store is not None: + try: + platform_enum = Platform(platform_name.lower()) + except (ValueError, KeyError): + platform_enum = None + if platform_enum is not None: + dest_source = SessionSource( + platform=platform_enum, + chat_id=str(chat_id), + chat_name=chat_name, + chat_type=chat_type, + user_id=str(user_id) if user_id else None, + thread_id=None, # flat — the whole-channel/DM session + ) + # Create the flat session row so the mirror has a target and the + # user's later plain reply joins the SAME session. + session_store.get_or_create_session(dest_source) + + from gateway.mirror import mirror_to_session + + ok = mirror_to_session( + platform_name, + str(chat_id), + f"[Cron delivery: {job.get('name') or job.get('id', 'cron')}]\n{text}", + source_label="cron", + thread_id=None, + user_id=str(user_id) if user_id else None, + role="user", + ) + if ok: + logger.info( + "Job '%s': seeded flat in_channel session on %s:%s (chat_type=%s)", + job.get("id", "?"), platform_name, chat_id, chat_type, + ) + return bool(ok) + except Exception as e: + logger.debug( + "Job '%s': seeding in_channel session failed for %s:%s: %s", + job.get("id", "?"), platform_name, chat_id, e, + ) + return False + + def _cron_job_origin_log_suffix(job: dict) -> str: """Return safe provenance details for security warnings about a cron job. @@ -1291,6 +1387,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option ) 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 @@ -1548,10 +1658,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}" diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 6a562ac66b4..d53209fd394 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -4168,7 +4168,7 @@ class TestCronContinuableSurfaceInChannel: mock_cfg.platforms = {Platform.SLACK: pconfig} return mock_cfg - def _run_inchannel_delivery(self, extra, adapter, *, mirror_ok=True): + 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.""" @@ -4194,7 +4194,9 @@ class TestCronContinuableSurfaceInChannel: "name": "Daily Brief", "deliver": "origin", # Channel origin: no thread_id (flat channel message scheduled it). - "origin": {"platform": "slack", "chat_id": "C123"}, + # 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, } @@ -4210,11 +4212,16 @@ class TestCronContinuableSurfaceInChannel: ) return open_thread_mock, mirror_mock - def _slack_adapter(self, supports_inchannel=True): + 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): @@ -4226,19 +4233,50 @@ class TestCronContinuableSurfaceInChannel: 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).""" + """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() - # 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_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.""" @@ -4273,3 +4311,81 @@ class TestCronContinuableSurfaceInChannel: ) 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() + diff --git a/tests/manual/cron_inchannel_e2e.py b/tests/manual/cron_inchannel_e2e.py index aaa525f191c..7d852ece5ac 100644 --- a/tests/manual/cron_inchannel_e2e.py +++ b/tests/manual/cron_inchannel_e2e.py @@ -1,174 +1,157 @@ """ -Offline E2E harness for continuable in-channel cron (specs/cron-inchannel-continuable). +Offline E2E 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)``: +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). - 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. +Two scenarios, each asserting the brief actually lands in the SAME session the +inbound reply resolves to: - 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. + 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. -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). + 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's code loads, not the editable -main-checkout install: +Run from INSIDE the worktree (so the worktree 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). +Uses a throwaway HERMES_HOME so it never touches ~/.hermes. No real names. """ -import asyncio +import os 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." +import tempfile +from pathlib import Path -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) +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) -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 +HOME = _fresh_home() - 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 +# 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 - captured = [] - a.handle_message = AsyncMock(side_effect=lambda e: captured.append(e)) +# 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" - 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", - } +BRIEF = "brief: PRs need review\n- Harden: session lifecycle teardown" - 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" +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", ) - return ("slack", src.chat_id, src.thread_id) + 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_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) + 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) - seed_key = leg1_delivery_seeds_flat_channel_session() - print(f"LEG 1 (delivery seed) → session key {seed_key}") + _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") - 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." + "\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)." )