feat(relay): flat in_channel continuable crons + block-formatting hints on the relay lane

docs/relay-connector-contract.md

@@ -52,6 +52,8 @@ JSON object. Source of truth: `gateway/relay/descriptor.py`.
 | `platform_hint` | string | no | System-prompt platform hint. |
 | `pii_safe` | bool | no | Redact PII in session descriptions. |
 | `supports_context` | bool | no | Whether the connector can supply surrounding channel/group **context** for an addressed turn on this platform (Model A on-demand history fetch — Discord/Slack/Matrix; Model B passive buffer — Telegram/Signal/WhatsApp). Default false ⇒ no `context` is attached to inbound events. See §3. |
+| `supports_inchannel_continuable` | bool | no | Whether the platform can host a **flat continuable cron surface** (native Slack's `cron_continuable_surface: in_channel`): the brief posts top-level in the channel/DM and a plain reply continues the job via the flat `(platform, chat_id, None)` session. Default false ⇒ the gateway's scheduler fails safe to thread mode (D6 gate), so an older connector keeps today's thread behavior. |
+| `supports_block_formatting` | bool | no | Whether this platform's sender renders **block-level formatting** from raw markdown when the gateway stamps `metadata.format_hints` on `send`/`edit` frames (Slack: native `markdown` block for tables/lists/code, mrkdwn text kept as fallback). Default false ⇒ the gateway never stamps hints, so an older connector never receives the metadata. |
 | `supported_ops` | string[] | no | Op-level capability discovery: the outbound op names the connector's sender for this platform actually implements (e.g. `["send", "edit", "typing", "follow_up", "get_chat_info"]`). Absent/empty ⇒ the connector predates the field and the gateway assumes the legacy op set (`send`/`edit`/`typing`/`follow_up`); a NEW op is used only when explicitly advertised. |
 
 Most fields are a projection of the gateway's existing `PlatformEntry`; the

gateway/mirror.py

@@ -30,13 +30,25 @@ def mirror_to_session(
     thread_id: Optional[str] = None,
     user_id: Optional[str] = None,
     role: str = "assistant",
+    session_id: Optional[str] = None,
 ) -> bool:
     """
     Append a delivery-mirror message to the target session's transcript.
 
     Finds the gateway session that matches the given platform + chat_id,
     then writes a mirror entry to both the JSONL transcript and SQLite DB.
 
+    ``session_id``: when the caller already KNOWS the exact session (e.g. the
+    cron in_channel seed, which just created the row via
+    ``get_or_create_session``), pass it to skip the origin-scan heuristics
+    entirely. ``_find_session_id`` matches by origin (chat_id + user
+    preference with a multi-candidate bail-out), which is correct for
+    "mirror into whatever conversation lives here" callers but WRONG for a
+    caller holding the precise target — on a populated chat (flat session +
+    N per-message thread sessions sharing one chat_id) the scan can refuse
+    to guess and silently drop the mirror (live failure, Alice 2026-08-19:
+    'in_channel seed did NOT land').
+
     ``role`` defaults to ``"assistant"`` — correct for the interactive
     ``send_message`` mirror, where the mirrored text is the agent's own
     outgoing reply (a genuine assistant turn). Callers mirroring text that is
@@ -52,15 +64,17 @@ def mirror_to_session(
     All errors are caught -- this is never fatal.
     """
     try:
-        session_id = _find_session_id(
-            platform,
-            str(chat_id),
-            thread_id=thread_id,
-            user_id=user_id,
-        )
         if not session_id:
-            logger.debug(
-                "Mirror: no session found for %s:%s:%s:%s",
+            session_id = _find_session_id(
+                platform,
+                str(chat_id),
+                thread_id=thread_id,
+                user_id=user_id,
+            )
+        if not session_id:
+            logger.warning(
+                "Mirror: no session found for %s:%s thread=%s user=%s "
+                "(explicit_id=none, origin-scan bailed)",
                 platform,
                 chat_id,
                 thread_id,
@@ -82,12 +96,17 @@ def mirror_to_session(
         return True
 
     except Exception as e:
-        logger.debug(
-            "Mirror failed for %s:%s:%s:%s: %s",
+        # WARNING with the exception: a silent mirror drop IS the cron
+        # continuation-amnesia bug (Alice 2026-08-19 — the seed's own
+        # deterministic session_id was in hand and the append STILL failed
+        # invisibly at debug level).
+        logger.warning(
+            "Mirror failed for %s:%s thread=%s user=%s session=%s: %s",
             platform,
             chat_id,
             thread_id,
             user_id,
+            session_id,
             e,
         )
         return False

gateway/relay/adapter.py

@@ -161,6 +161,11 @@ def __init__(
         # _capture_scope / send.
         self._platform_by_chat: Dict[str, str] = {}
         self.supports_code_blocks = descriptor.markdown_dialect not in ("", "plain")
+        # Cron flat continuable surface — descriptor-advertised (see
+        # _apply_descriptor; same bit, constructor path).
+        self.supports_inchannel_continuable = bool(
+            getattr(descriptor, "supports_inchannel_continuable", False)
+        )
         # Phase 7 Unit 7d-B: watches the transport for a terminal auth revocation
         # (a 4401 close after a successful handshake = the operator opted this
         # instance out of the relay). On revocation we surface a clean,
@@ -892,6 +897,13 @@ def _apply_descriptor(self, descriptor: CapabilityDescriptor) -> None:
         self.descriptor = descriptor
         self.MAX_MESSAGE_LENGTH = descriptor.max_message_length
         self.supports_code_blocks = descriptor.markdown_dialect not in ("", "plain")
+        # Cron in_channel continuable surface (D6 gate in cron/scheduler.py):
+        # the scheduler reads this off the adapter; the connector advertises it
+        # per platform at handshake. Class default is False (BasePlatformAdapter),
+        # so only an explicit descriptor bit turns the flat surface on.
+        self.supports_inchannel_continuable = bool(
+            getattr(descriptor, "supports_inchannel_continuable", False)
+        )
 
     async def _on_inbound(self, event) -> None:
         """Bridge a connector-delivered MessageEvent into the normal adapter path."""
@@ -1250,6 +1262,36 @@ def _platform_is_fronted(self, platform: str) -> bool:
         """Backward-compatible internal alias for follow-up routing."""
         return self.fronts_platform(platform)
 
+    def supports_inchannel_continuable_for_platform(self, platform: Any) -> bool:
+        """Whether ONE fronted logical platform can host the flat continuable
+        cron surface (the D6 gate in cron/scheduler.py).
+
+        The scalar ``supports_inchannel_continuable`` carries only the PRIMARY
+        identity's bit, but one RelayAdapter fronts N platforms and the
+        connector advertises the capability per platform at handshake. On a
+        multi-platform relay the scalar both leaks the primary's True onto
+        platforms whose own descriptor never advertised it and suppresses a
+        non-primary platform's advertised True. Resolve the platform's own
+        negotiated descriptor off the transport; fall back to the scalar only
+        when the per-platform descriptor is unavailable (single-platform
+        transport, or a transport predating descriptor_for_platform).
+        """
+        platform_value = str(getattr(platform, "value", platform) or "")
+        if platform_value and self._transport is not None:
+            resolve = getattr(self._transport, "descriptor_for_platform", None)
+            if callable(resolve):
+                try:
+                    per_platform = resolve(platform_value)
+                except Exception:  # noqa: BLE001 - capability lookup must never break delivery
+                    per_platform = None
+                if per_platform is not None:
+                    return bool(
+                        getattr(
+                            per_platform, "supports_inchannel_continuable", False
+                        )
+                    )
+        return bool(self.supports_inchannel_continuable)
+
     async def on_interrupt(self, session_key: str, chat_id: str) -> None:
         """Bridge a connector-delivered /stop into the adapter's interrupt path.
 
@@ -1637,7 +1679,17 @@ async def send_for_platform(
                 "chat_id": chat_id,
                 "content": content,
                 "reply_to": reply_to,
-                "metadata": self._with_scope(chat_id, _sfp_metadata),
+                # format_hints on the explicit-platform lane too: this is the
+                # scheduled/cron delivery path — the in_channel brief itself —
+                # and it must render blocks exactly like an interactive send.
+                # Stamps _sfp_metadata (the interim-marker-stripped copy, per
+                # the seal path above), composing both sides of the merge.
+                "metadata": self._with_scope(
+                    chat_id,
+                    self._with_format_hints_for_platform(
+                        str(platform_value), _sfp_metadata
+                    ),
+                ),
             },
             platform=str(platform_value),
         )
@@ -1648,6 +1700,101 @@ async def send_for_platform(
             raw_response=result,
         )
 
+    def _format_hints(
+        self, descriptor: Optional[CapabilityDescriptor], platform: Optional[str]
+    ) -> Optional[Dict[str, bool]]:
+        """Block-formatting hints for one outbound text frame, or None.
+
+        Native Slack reads ``platforms.slack.extra.rich_blocks`` /
+        ``markdown_blocks`` and renders Block Kit locally; on the relay lane
+        the CONNECTOR owns the platform API call, so the gateway can only
+        signal intent. Hints are stamped ONLY when (a) the DESTINATION
+        platform's negotiated descriptor advertises
+        ``supports_block_formatting`` — an old connector never receives dead
+        metadata — and (b) the operator enabled at least one knob under the
+        relay's per-logical-platform sub-block
+        (``platforms.relay.extra.<platform>.rich_blocks`` /
+        ``markdown_blocks``, same seam and same _coerce_flag semantics as
+        reply_in_thread). Both knobs default OFF, matching native's opt-in
+        posture.
+
+        ``descriptor``/``platform`` are the DESTINATION's, not the adapter's
+        scalar primary identity: one RelayAdapter fronts N platforms, and
+        gating on the primary descriptor both leaked hints onto platforms
+        that never advertised the bit (Slack-primary, Discord chat) and
+        suppressed them for platforms that did (Discord-primary, Slack chat).
+        Same seam as ``_descriptor_for_chat`` / max_message_length.
+        """
+        if descriptor is None or not getattr(
+            descriptor, "supports_block_formatting", False
+        ):
+            return None
+        try:
+            extra = getattr(self.config, "extra", None) or {}
+            sub = extra.get(str(platform or "").lower())
+            knob_src = sub if isinstance(sub, dict) else extra
+        except Exception:  # noqa: BLE001 - config shape is operator-owned
+            return None
+        hints: Dict[str, bool] = {}
+        for knob in ("rich_blocks", "markdown_blocks"):
+            if self._coerce_flag(knob_src.get(knob), False):
+                hints[knob] = True
+        return hints or None
+
+    def _with_format_hints_for_chat(
+        self, chat_id: str, metadata: Optional[Dict[str, Any]]
+    ) -> Optional[Dict[str, Any]]:
+        """Metadata with ``format_hints`` stamped for a chat-addressed send.
+
+        Resolves the chat's platform from what we saw inbound
+        (``_platform_by_chat``) and that platform's negotiated descriptor
+        (``_descriptor_for_chat``) — falling back to the primary identity for
+        chats we never saw inbound, matching every other per-chat capability.
+        """
+        platform = self._platform_by_chat.get(str(chat_id)) or getattr(
+            self.descriptor, "platform", None
+        )
+        hints = self._format_hints(self._descriptor_for_chat(chat_id), platform)
+        if not hints:
+            return metadata
+        merged = dict(metadata or {})
+        merged.setdefault("format_hints", hints)
+        return merged
+
+    def _with_format_hints_for_platform(
+        self, platform_value: str, metadata: Optional[Dict[str, Any]]
+    ) -> Optional[Dict[str, Any]]:
+        """Metadata with ``format_hints`` stamped for an explicit-platform send.
+
+        ``send_for_platform`` is the scheduled/persisted-home lane — the cron
+        delivery path, i.e. the flagship consumer of the in_channel brief —
+        and it has no inbound event to populate ``_platform_by_chat``, so the
+        destination platform is the caller-supplied logical platform.
+        Resolves that platform's negotiated descriptor off the transport;
+        falls back to the scalar descriptor only when it IS that platform's
+        (fail closed: never stamp from another platform's capability bit).
+        """
+        descriptor: Optional[CapabilityDescriptor] = None
+        if self._transport is not None:
+            resolve = getattr(self._transport, "descriptor_for_platform", None)
+            if callable(resolve):
+                try:
+                    descriptor = cast(
+                        Optional[CapabilityDescriptor], resolve(str(platform_value))
+                    )
+                except Exception:  # noqa: BLE001 - capability lookup must never break a send
+                    descriptor = None
+        if descriptor is None and getattr(
+            self.descriptor, "platform", None
+        ) == str(platform_value):
+            descriptor = self.descriptor
+        hints = self._format_hints(descriptor, str(platform_value))
+        if not hints:
+            return metadata
+        merged = dict(metadata or {})
+        merged.setdefault("format_hints", hints)
+        return merged
+
     async def send(
         self,
         chat_id: str,
@@ -1715,7 +1862,9 @@ async def send(
                 "chat_id": chat_id,
                 "content": content,
                 "reply_to": effective_reply_to,
-                "metadata": self._with_scope(chat_id, send_metadata),
+                "metadata": self._with_scope(
+                    chat_id, self._with_format_hints_for_chat(chat_id, send_metadata)
+                ),
             },
             platform=self._platform_by_chat.get(str(chat_id)),
         )
@@ -1953,7 +2102,13 @@ async def edit_message(
                 "chat_id": chat_id,
                 "message_id": message_id,
                 "content": content,
-                "metadata": self._with_scope(chat_id, metadata),
+                # Same format_hints as send: a streamed reply's FINAL edit is
+                # the frame that carries the finished markdown, so the edit
+                # lane must signal block rendering too or streams would seal
+                # as plain text (boundary rule: every text egress lane).
+                "metadata": self._with_scope(
+                    chat_id, self._with_format_hints_for_chat(chat_id, metadata)
+                ),
             },
             platform=self._platform_by_chat.get(str(chat_id)),
         )

gateway/relay/descriptor.py

@@ -64,6 +64,23 @@ class CapabilityDescriptor:
     # "no context" — additive within contract_version 1. from_json filters
     # unknown keys, so a connector sending this to an older gateway is safe too.
     supports_context: bool = False
+    # Whether the connector's platform can host a FLAT continuable cron
+    # surface (native Slack's ``cron_continuable_surface: in_channel``): the
+    # brief posts top-level in the channel/DM and a plain reply continues the
+    # job via the flat ``(platform, chat_id, None)`` session. The scheduler
+    # fails safe to thread mode when False (D6 gate), so an older connector
+    # that never sends this keeps today's thread behavior — additive within
+    # contract_version 1.
+    supports_inchannel_continuable: bool = False
+    # Whether the connector's platform sender can render block-level
+    # formatting from raw markdown (Slack: rich_text lists, Block Kit
+    # tables/markdown blocks). When True AND the operator enables the
+    # rich_blocks/markdown_blocks knobs, the gateway stamps ``format_hints``
+    # into outbound send/edit metadata; the connector renders blocks and
+    # keeps the plain text as fallback. Default False — old connectors never
+    # receive hints, old gateways never send them. Additive within
+    # contract_version 1.
+    supports_block_formatting: bool = False
     # Op-level capability discovery (Phase 1 parity): the outbound op names the
     # connector's sender for this platform actually implements (e.g.
     # ["send", "edit", "typing", "follow_up", "get_chat_info"]). Empty tuple =