diff --git a/cron/scheduler.py b/cron/scheduler.py
index 998af72d7727..3427c75b5b62 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -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,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
         # continues with full context. On DM-only platforms (WhatsApp/Signal)
         # create_handoff_thread returns None and we fall back to mirroring into
         # the origin DM session (handled after delivery). Cf. _process_handoff.
+        #
+        # in_channel surface (D2): SKIP thread creation entirely — leave
+        # thread_id=None so the delivery posts flat, then
+        # ``_seed_cron_channel_session`` (below) CREATES the shared-channel
+        # session and mirrors the brief into it. The shipped mirror alone is
+        # NOT enough here: ``mirror_to_session`` only APPENDS to an existing
+        # session and a flat ``(platform, chat_id, None)`` row is otherwise
+        # absent for a ``chat_postMessage`` delivery, so the seed must create
+        # the row first (F5).
         thread_seeded = False
         opened_thread_id: Optional[str] = None
         if (
             mirror_this_target
+            and not in_channel_surface
             and runtime_adapter is not None
             and loop is not None
             and not thread_id  # never override an explicit origin thread/topic
@@ -1454,10 +1604,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/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 b6834a669c90..1025964dc43b 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -2307,6 +2307,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
diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py
index 521d82f4016a..af86ff4f2753 100644
--- a/plugins/platforms/slack/adapter.py
+++ b/plugins/platforms/slack/adapter.py
@@ -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 @@ 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")
@@ -1562,6 +1571,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,
@@ -2097,10 +2162,10 @@ async def send_image(
 
             async def _ssrf_redirect_guard(response):
                 """Re-check redirect targets so public URLs cannot bounce into private IPs."""
-                if response.is_redirect and response.next_request:
-                    redirect_url = str(response.next_request.url)
-                    if not is_safe_url(redirect_url):
-                        raise ValueError("Blocked redirect to private/internal address")
+                from tools.url_safety import redirect_target_from_response
+                redirect_url = redirect_target_from_response(response)
+                if redirect_url and not is_safe_url(redirect_url):
+                    raise ValueError("Blocked redirect to private/internal address")
 
             # Download the image first
             async with httpx.AsyncClient(
diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py
index 935533b11b97..01bae7ed2d4b 100644
--- a/tests/cron/test_scheduler.py
+++ b/tests/cron/test_scheduler.py
@@ -1,5 +1,6 @@
 """Tests for cron/scheduler.py — origin resolution, delivery routing, and error logging."""
 
+import contextlib
 import json
 import logging
 import os
@@ -966,7 +967,8 @@ def test_run_job_passes_session_db_and_cron_platform(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1012,7 +1014,8 @@ def test_run_job_suppresses_empty_turn_explainer(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1055,7 +1058,8 @@ def test_run_job_real_report_on_empty_reason_still_delivers(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1095,7 +1099,8 @@ def test_run_job_titles_cron_session_from_job_not_important_hint(self, tmp_path)
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1132,7 +1137,8 @@ def test_run_job_closes_agent_on_failure_to_prevent_fd_leak(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1169,7 +1175,8 @@ def test_run_job_reaps_stale_auxiliary_clients_per_tick(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1191,13 +1198,28 @@ def test_run_job_reaps_stale_auxiliary_clients_per_tick(self, tmp_path):
         assert success is True
         cleanup_mock.assert_called_once()
 
-    def _make_run_job_patches(self, tmp_path):
-        """Common patches for run_job tests."""
+    @contextlib.contextmanager
+    def _run_job_patches(self, tmp_path, extra=()):
+        """Apply every patch run_job tests need, as one bundle.
+
+        Yields ``(fake_db, mock_agent_cls)``. Using an ExitStack that enters
+        the whole list means a caller can never silently drop a patch by
+        index — the previous positional-list form let a seam split shift
+        ``resolve_runtime_provider`` off the end of the applied slice, so the
+        real resolver ran and (only on a dev machine with ambient creds) hid
+        an auth failure that CI then caught. Every test enters all patches.
+
+        ``extra`` is an iterable of additional context managers (e.g. a
+        per-test ``_get_platform_tools`` patch) entered alongside the base set.
+        """
         fake_db = MagicMock()
-        return fake_db, [
+        mock_agent = MagicMock()
+        mock_agent.run_conversation.return_value = {"final_response": "ok"}
+        base = [
             patch("cron.scheduler._hermes_home", tmp_path),
             patch("cron.scheduler._resolve_origin", return_value=None),
-            patch("dotenv.load_dotenv"),
+            patch("hermes_cli.env_loader.load_hermes_dotenv"),
+            patch("hermes_cli.env_loader.reset_secret_source_cache"),
             patch("hermes_state.SessionDB", return_value=fake_db),
             patch(
                 "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1208,7 +1230,14 @@ def _make_run_job_patches(self, tmp_path):
                     "api_mode": "chat_completions",
                 },
             ),
+            patch("run_agent.AIAgent", return_value=mock_agent),
         ]
+        with contextlib.ExitStack() as stack:
+            entered = [stack.enter_context(cm) for cm in base]
+            for cm in extra:
+                stack.enter_context(cm)
+            mock_agent_cls = entered[-1]  # the AIAgent patch
+            yield fake_db, mock_agent_cls
 
     def test_run_job_passes_enabled_toolsets_to_agent(self, tmp_path):
         job = {
@@ -1217,12 +1246,7 @@ def test_run_job_passes_enabled_toolsets_to_agent(self, tmp_path):
             "prompt": "hello",
             "enabled_toolsets": ["web", "terminal", "file"],
         }
-        fake_db, patches = self._make_run_job_patches(tmp_path)
-        with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patch("run_agent.AIAgent") as mock_agent_cls:
-            mock_agent = MagicMock()
-            mock_agent.run_conversation.return_value = {"final_response": "ok"}
-            mock_agent_cls.return_value = mock_agent
+        with self._run_job_patches(tmp_path) as (_fake_db, mock_agent_cls):
             run_job(job)
 
         kwargs = mock_agent_cls.call_args.kwargs
@@ -1251,12 +1275,7 @@ def test_run_job_disabled_toolsets_layer_user_config_on_baseline(self, tmp_path)
             "prompt": "hello",
             "enabled_toolsets": ["web", "terminal", "file"],
         }
-        fake_db, patches = self._make_run_job_patches(tmp_path)
-        with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patch("run_agent.AIAgent") as mock_agent_cls:
-            mock_agent = MagicMock()
-            mock_agent.run_conversation.return_value = {"final_response": "ok"}
-            mock_agent_cls.return_value = mock_agent
+        with self._run_job_patches(tmp_path) as (_fake_db, mock_agent_cls):
             run_job(job)
 
         kwargs = mock_agent_cls.call_args.kwargs
@@ -1278,12 +1297,7 @@ def test_run_job_enabled_toolsets_resolves_from_platform_config_when_not_set(sel
             "name": "test",
             "prompt": "hello",
         }
-        fake_db, patches = self._make_run_job_patches(tmp_path)
-        with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patch("run_agent.AIAgent") as mock_agent_cls:
-            mock_agent = MagicMock()
-            mock_agent.run_conversation.return_value = {"final_response": "ok"}
-            mock_agent_cls.return_value = mock_agent
+        with self._run_job_patches(tmp_path) as (_fake_db, mock_agent_cls):
             run_job(job)
 
         kwargs = mock_agent_cls.call_args.kwargs
@@ -1304,18 +1318,10 @@ def test_run_job_per_job_toolsets_win_over_platform_config(self, tmp_path):
             "prompt": "hello",
             "enabled_toolsets": ["terminal"],
         }
-        fake_db, patches = self._make_run_job_patches(tmp_path)
         # Even if the user has ``hermes tools`` configured to enable web+file
         # for cron, the per-job override wins.
-        with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patch("run_agent.AIAgent") as mock_agent_cls, \
-             patch(
-                 "hermes_cli.tools_config._get_platform_tools",
-                 return_value={"web", "file"},
-             ):
-            mock_agent = MagicMock()
-            mock_agent.run_conversation.return_value = {"final_response": "ok"}
-            mock_agent_cls.return_value = mock_agent
+        extra = [patch("hermes_cli.tools_config._get_platform_tools", return_value={"web", "file"})]
+        with self._run_job_patches(tmp_path, extra=extra) as (_fake_db, mock_agent_cls):
             run_job(job)
 
         kwargs = mock_agent_cls.call_args.kwargs
@@ -1336,7 +1342,8 @@ def test_run_job_empty_response_returns_empty_not_placeholder(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1412,7 +1419,8 @@ def test_run_job_treats_agent_failure_flag_as_failure(
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1451,7 +1459,8 @@ def test_run_job_completed_true_without_failed_flag_succeeds(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1493,7 +1502,8 @@ def test_run_job_delivers_max_iteration_fallback_summary(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -1613,6 +1623,53 @@ def run_conversation(self, *args, **kwargs):
         assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None
         fake_db.close.assert_called_once()
 
+    def test_run_job_resets_secret_source_cache_before_reload(self, tmp_path, monkeypatch):
+        """Each run must clear the secret-source cache before re-reading the
+        env, so a long-running gateway re-resolves Bitwarden/BSM-backed secrets
+        instead of leaving the startup .env placeholder in place (#33465).
+
+        A bare ``load_dotenv`` re-load can't do this: startup already recorded
+        this HERMES_HOME in ``_APPLIED_HOMES``, so the external-secret pull
+        no-ops and only the placeholder is re-applied. The scheduler must call
+        ``reset_secret_source_cache()`` (forcing the re-pull) and route through
+        ``load_hermes_dotenv`` (which then re-applies external secret sources).
+        """
+        job = {"id": "bsm-job", "name": "bsm", "prompt": "hello"}
+        fake_db = MagicMock()
+        call_order = []
+
+        def _record_reset():
+            call_order.append("reset")
+
+        def _record_load(*args, **kwargs):
+            call_order.append("load")
+            return []
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache", _record_reset), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv", _record_load), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch(
+                 "hermes_cli.runtime_provider.resolve_runtime_provider",
+                 return_value={
+                     "api_key": "***",
+                     "base_url": "https://example.invalid/v1",
+                     "provider": "openrouter",
+                     "api_mode": "chat_completions",
+                 },
+             ), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent.run_conversation.return_value = {"final_response": "ok"}
+            mock_agent_cls.return_value = mock_agent
+            success, _output, _final, error = run_job(job)
+
+        assert success is True
+        assert error is None
+        # reset MUST precede the reload, else _APPLIED_HOMES no-ops the re-pull.
+        assert call_order[:2] == ["reset", "load"], call_order
+
     def test_run_job_clears_stale_auto_delivery_thread_id_between_jobs(self, tmp_path, monkeypatch):
         jobs = [
             {
@@ -1709,7 +1766,8 @@ def test_bad_config_yaml_is_logged(self, caplog, tmp_path):
         # (>30s wall clock) under load. See PR #33661 follow-up.
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value={"provider": "openrouter", "api_key": "x",
                                  "base_url": "https://example.invalid",
@@ -1743,7 +1801,8 @@ def test_bad_prefill_messages_is_logged(self, caplog, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value={"provider": "openrouter", "api_key": "x",
                                  "base_url": "https://example.invalid",
@@ -1781,7 +1840,8 @@ def test_model_env_ref_in_config_yaml_is_expanded(self, tmp_path, monkeypatch):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1814,7 +1874,8 @@ def test_legacy_agent_prefill_messages_file_is_loaded(self, tmp_path, monkeypatc
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1844,7 +1905,8 @@ def test_fallback_model_env_ref_in_config_yaml_is_expanded(self, tmp_path, monke
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1863,6 +1925,36 @@ def test_fallback_model_env_ref_in_config_yaml_is_expanded(self, tmp_path, monke
             "config.yaml ${VAR} in fallback_providers was not expanded."
         )
 
+    def test_fallback_chain_merges_providers_and_legacy_model(self, tmp_path, monkeypatch):
+        """Cron uses get_fallback_chain so legacy fallback_model is not dropped."""
+        (tmp_path / "config.yaml").write_text(
+            "fallback_providers:\n"
+            "  - provider: openrouter\n"
+            "    model: gpt-4o-mini\n"
+            "fallback_model:\n"
+            "  provider: anthropic\n"
+            "  model: claude-sonnet-4-6\n"
+        )
+
+        job = {"id": "fb-merge", "name": "fallback merge", "prompt": "hi"}
+        fake_db = MagicMock()
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch("hermes_cli.runtime_provider.resolve_runtime_provider",
+                   return_value=self._RUNTIME), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent.run_conversation.return_value = {"final_response": "ok"}
+            mock_agent_cls.return_value = mock_agent
+            run_job(job)
+
+        fb = mock_agent_cls.call_args.kwargs.get("fallback_model") or []
+        models = [e.get("model") for e in fb if isinstance(e, dict)]
+        assert models == ["gpt-4o-mini", "claude-sonnet-4-6"]
+
     def test_unexpanded_ref_passthrough_when_var_unset(self, tmp_path, monkeypatch):
         """When the env var is not set, the literal ${VAR} is kept verbatim (not crashed)."""
         (tmp_path / "config.yaml").write_text("model: ${_HERMES_TEST_CRON_UNSET_VAR}\n")
@@ -1873,7 +1965,8 @@ def test_unexpanded_ref_passthrough_when_var_unset(self, tmp_path, monkeypatch):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1917,7 +2010,8 @@ def test_null_job_model_falls_back_to_env(self, tmp_path, monkeypatch):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1941,7 +2035,8 @@ def test_null_job_model_falls_back_to_config_default(self, tmp_path, monkeypatch
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1973,7 +2068,8 @@ def test_explicit_null_model_block_in_config_does_not_overwrite_env(self, tmp_pa
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -1996,7 +2092,8 @@ def test_no_model_anywhere_fails_with_actionable_error(self, tmp_path, monkeypat
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -2025,7 +2122,8 @@ def test_job_model_update_takes_effect_on_next_run(self, tmp_path, monkeypatch):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -2051,7 +2149,8 @@ def test_config_model_as_plain_string(self, tmp_path, monkeypatch):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -2081,7 +2180,8 @@ def test_config_model_alias_key_resolves(self, tmp_path, monkeypatch):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -2105,7 +2205,8 @@ def test_corrupt_config_yaml_does_not_crash_with_job_model(self, tmp_path, monke
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch("hermes_cli.runtime_provider.resolve_runtime_provider",
                    return_value=self._RUNTIME), \
@@ -2147,7 +2248,8 @@ def _run_conversation(prompt):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -2207,7 +2309,8 @@ def _run_conversation(prompt):
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
              patch("tools.credential_files._resolve_hermes_home", return_value=tmp_path), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -2245,7 +2348,8 @@ def test_run_job_loads_skill_and_disables_recursive_cron_tools(self, tmp_path):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -2291,7 +2395,8 @@ def _skill_view(name):
 
         with patch("cron.scheduler._hermes_home", tmp_path), \
              patch("cron.scheduler._resolve_origin", return_value=None), \
-             patch("dotenv.load_dotenv"), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
              patch("hermes_state.SessionDB", return_value=fake_db), \
              patch(
                  "hermes_cli.runtime_provider.resolve_runtime_provider",
@@ -2466,6 +2571,46 @@ def test_whitespace_only_response_is_marked_failed_not_delivered(self):
         )
 
 
+class TestOneShotDispatchClaim:
+    """run_one_job must claim a finite one-shot's dispatch BEFORE run_job so a
+    tick that dies mid-execution can't re-fire it forever (issue #38758)."""
+
+    def _oneshot(self):
+        return {
+            "id": "monitor-job",
+            "name": "monitor",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "123"},
+            "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00+00:00"},
+            "repeat": {"times": 1, "completed": 0},
+        }
+
+    def test_claim_runs_before_run_job(self):
+        order = []
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \
+             patch("cron.scheduler.claim_dispatch", side_effect=lambda _id: order.append("claim") or True), \
+             patch("cron.scheduler.run_job", side_effect=lambda _j: order.append("run") or (True, "# out", "ok", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result"), \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        assert order == ["claim", "run"]  # claim strictly before side effect
+
+    def test_refused_claim_skips_run_job(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \
+             patch("cron.scheduler.claim_dispatch", return_value=False), \
+             patch("cron.scheduler.run_job") as run_mock, \
+             patch("cron.scheduler.save_job_output"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run") as mark_mock:
+            from cron.scheduler import tick
+            tick(verbose=False)
+        run_mock.assert_not_called()
+        deliver_mock.assert_not_called()
+        mark_mock.assert_not_called()
+
+
 class TestBuildJobPromptSilentHint:
     """Verify _build_job_prompt always injects [SILENT] guidance."""
 
@@ -3997,3 +4142,252 @@ 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), then ``_seed_cron_channel_session`` CREATES
+    the flat shared-channel session and mirrors the brief into it (the shipped
+    mirror only APPENDS to an existing session, and the flat channel row is
+    otherwise absent for a chat_postMessage delivery).
+    """
+
+    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()
+
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_dm_e2e.py b/tests/manual/cron_inchannel_dm_e2e.py
new file mode 100644
index 000000000000..16273ce1a520
--- /dev/null
+++ b/tests/manual/cron_inchannel_dm_e2e.py
@@ -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()
diff --git a/tests/manual/cron_inchannel_e2e.py b/tests/manual/cron_inchannel_e2e.py
new file mode 100644
index 000000000000..7d852ece5acb
--- /dev/null
+++ b/tests/manual/cron_inchannel_e2e.py
@@ -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()
diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md
index cfbe48b40f05..a65a4bca1e78 100644
--- a/website/docs/user-guide/features/cron.md
+++ b/website/docs/user-guide/features/cron.md
@@ -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.
diff --git a/website/docs/user-guide/messaging/slack.md b/website/docs/user-guide/messaging/slack.md
index 34827cf79d0e..fbfdca794118 100644
--- a/website/docs/user-guide/messaging/slack.md
+++ b/website/docs/user-guide/messaging/slack.md
@@ -352,6 +352,13 @@ platforms:
       # Tables exceeding Slack's limits (100 rows / 20 cols / 10k chars)
       # gracefully fall back to aligned monospace.
       rich_blocks: 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 |
@@ -360,6 +367,7 @@ platforms:
 | `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.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. |
+| `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..e543f8cfcb22 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,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/`），但不会向投递目标发送任何消息。
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 02ee4c9cf0cf..9ebfb0998c3b 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
@@ -306,6 +306,13 @@ platforms:
       # 纯文本回退内容，用于通知和无障碍访问。超出 Slack 限制
       # （100 行 / 20 列 / 1 万字符）的表格会优雅地回退为对齐的等宽文本。
       rich_blocks: false
+
+      # 可继续 cron 任务的投递方式（默认："thread"）。
+      # "in_channel" 将可继续的 cron 任务直接平铺投递到频道中
+      # （不新建话题）；需与 reply_in_thread: false（及
+      # require_mention: false）搭配，纯文本回复即可继续任务。
+      # 详见 cron 指南 →“平铺频道内继续”。
+      cron_continuable_surface: thread
 ```
 
 | 键 | 默认值 | 描述 |
@@ -314,6 +321,7 @@ platforms:
 | `platforms.slack.extra.reply_in_thread` | `true` | 为 `false` 时，频道消息直接回复而非话题。已在话题中的消息仍在话题中回复。 |
 | `platforms.slack.extra.reply_broadcast` | `false` | 为 `true` 时，话题回复也会发布到主频道。仅广播第一个分块。 |
 | `platforms.slack.extra.rich_blocks` | `false` | 为 `true` 时，Agent 消息会渲染为 [Block Kit](https://docs.slack.dev/block-kit/) 区块（标题、分隔线、真正的嵌套列表以及原生表格）。始终附带纯文本回退。超出 Slack 限制的表格会回退为对齐的等宽文本。无需重新安装应用——这仅是发送端的改动。 |
+| `platforms.slack.extra.cron_continuable_surface` | `"thread"` | [可继续 cron 任务](../features/cron.md)的投递方式。`"thread"` 为每次投递新建专用话题（默认）；`"in_channel"` 直接平铺投递到频道时间线。使用 `in_channel` 时需搭配 `reply_in_thread: false`（及 `require_mention: false`），纯文本回复即可继续任务。 |
 
 ### 会话隔离