Merge origin/main into fix/cron-stream-id

maintainer edit: extract cron job execution, streaming buffering, notification gating, and turn_end emission into nanobot.cron.executor so commands.py only wires gateway dependencies.

nanobot/cli/commands.py

@@ -878,13 +878,12 @@ def _run_gateway(
     health_server_enabled: bool = True,
 ) -> None:
     """Shared gateway runtime; ``open_browser_url`` opens a tab once channels are up."""
-    from nanobot.agent.tools.cron import CronTool
     from nanobot.agent.tools.message import MessageTool
     from nanobot.bus.queue import MessageBus
     from nanobot.bus.runtime_events import RuntimeEventBus
     from nanobot.channels.manager import ChannelManager
+    from nanobot.cron.executor import CronJobExecutor
     from nanobot.cron.service import CronService
-    from nanobot.cron.types import CronJob
     from nanobot.providers.factory import build_provider_snapshot, load_provider_snapshot
     from nanobot.providers.image_generation import image_gen_provider_configs
     from nanobot.session.manager import SessionManager
@@ -976,174 +975,44 @@ def _run_gateway(
     if isinstance(message_tool, MessageTool):
         message_tool.set_send_callback(_deliver_to_channel)
 
-    # Set cron callback (needs agent)
-    async def on_cron_job(job: CronJob) -> str | None:
-        """Execute a cron job through the agent."""
-        async def _silent(*_args, **_kwargs):
-            pass
+    hb_cfg = config.gateway.heartbeat
 
-        # Dream is an internal job — run directly, not through the agent loop.
-        if job.name == "dream":
-            from nanobot.agent.memory import MemoryStore
-
-            dream_session_key = MemoryStore.dream_session_key
-            build_dream_commit_message = MemoryStore.build_dream_commit_message
-            prune_dream_sessions = MemoryStore.prune_dream_sessions
-
-            store = agent.context.memory
-            resp = None
+    def _get_channel(channel_name: str) -> Any | None:
         try:
-                result = store.build_dream_prompt()
-                if result is None:
-                    logger.info("Dream: nothing to process")
-                    return None
-                prompt, last_cursor = result
-                key = dream_session_key()
-                resp = await agent.process_direct(
-                    prompt,
-                    session_key=key,
-                    ephemeral=True,
-                    tools=store.build_dream_tools(),
-                    on_progress=_silent,
-                )
-                if MemoryStore.dream_run_completed(resp):
-                    store.set_last_dream_cursor(last_cursor)
-                    logger.info("Dream cron job completed, cursor advanced to {}", last_cursor)
-                else:
-                    logger.warning(
-                        "Dream cron job did not complete; cursor remains at {}",
-                        store.get_last_dream_cursor(),
-                    )
-            except Exception:
-                logger.exception("Dream cron job failed")
-            finally:
-                if store.git.is_initialized():
-                    msg = build_dream_commit_message(
-                        "dream: periodic memory consolidation", resp,
-                    )
-                    sha = store.git.auto_commit(msg)
-                    if sha:
-                        logger.info("Dream commit: {}", sha)
-                store.compact_history()
-                prune_dream_sessions(agent.sessions.sessions_dir)
+            return channels.channels.get(channel_name)
+        except NameError:
             return None
 
-        # Heartbeat is a system job that checks HEARTBEAT.md for active tasks.
-        if job.name == "heartbeat":
-            heartbeat_file = config.workspace_path / "HEARTBEAT.md"
+    def _pick_heartbeat_target() -> tuple[str, str]:
+        """Pick a routable channel/chat target for heartbeat-triggered messages."""
         try:
-                content = heartbeat_file.read_text(encoding="utf-8")
-            except OSError:
-                logger.debug("Heartbeat: HEARTBEAT.md missing")
-                return None
-            if not _heartbeat_has_active_tasks(content):
-                logger.debug("Heartbeat: HEARTBEAT.md has no active tasks")
-                return None
+            enabled = set(channels.enabled_channels)
+        except NameError:
+            return "cli", "direct"
+        for item in session_manager.list_sessions():
+            key = item.get("key") or ""
+            if ":" not in key:
+                continue
+            channel, chat_id = key.split(":", 1)
+            if channel in {"cli", "system"}:
+                continue
+            if channel in enabled and chat_id:
+                return channel, chat_id
+        return "cli", "direct"
 
-            channel, chat_id = _pick_heartbeat_target()
-            if channel == "cli":
-                return None
-
-            prompt = (
-                _HEARTBEAT_PREAMBLE
-                + f"Review the following HEARTBEAT.md and report any active tasks:\n\n{content}"
+    cron_executor = CronJobExecutor(
+        agent=agent,
+        bus=bus,
+        deliver_to_channel=_deliver_to_channel,
+        get_channel=_get_channel,
+        evaluate_response=evaluate_response,
+        heartbeat_workspace=config.workspace_path,
+        heartbeat_preamble=_HEARTBEAT_PREAMBLE,
+        heartbeat_has_active_tasks=_heartbeat_has_active_tasks,
+        pick_heartbeat_target=_pick_heartbeat_target,
+        heartbeat_keep_recent_messages=hb_cfg.keep_recent_messages,
     )
-
-            # Internal check: funnel all output through the post-run gate so the
-            # turn can't deliver directly via the message tool and skip it.
-            suppress_token = None
-            if isinstance(message_tool, MessageTool):
-                suppress_token = message_tool.set_suppress_delivery(True)
-            try:
-                resp = await agent.process_direct(
-                    prompt,
-                    session_key="heartbeat",
-                    channel=channel,
-                    chat_id=chat_id,
-                    on_progress=_silent,
-                )
-            finally:
-                if isinstance(message_tool, MessageTool) and suppress_token is not None:
-                    message_tool.reset_suppress_delivery(suppress_token)
-            response = resp.content if resp else ""
-
-            # Keep a small tail of heartbeat history so the loop stays bounded.
-            session = agent.sessions.get_or_create("heartbeat")
-            session.retain_recent_legal_suffix(hb_cfg.keep_recent_messages)
-            agent.sessions.save(session)
-
-            if not response:
-                return None
-
-            # Fail closed: stay silent on evaluator failure instead of notifying.
-            should_notify = await evaluate_response(
-                response, prompt, agent.provider, agent.model,
-                default_notify=False,
-            )
-            if should_notify:
-                logger.info("Heartbeat: completed, delivering response")
-                await _deliver_to_channel(
-                    OutboundMessage(channel=channel, chat_id=chat_id, content=response),
-                    record=True,
-                )
-            else:
-                logger.info("Heartbeat: silenced by post-run evaluation")
-            return response
-
-        reminder_note = (
-            "The scheduled time has arrived. Deliver this reminder to the user now, "
-            "as a brief and natural message in their language. Speak directly to them — "
-            "do not narrate progress, summarize, include user IDs, or add status reports "
-            "like 'Done' or 'Reminded'.\n\n"
-            f"Reminder: {job.payload.message}"
-        )
-
-        cron_tool = agent.tools.get("cron")
-        cron_token = None
-        if isinstance(cron_tool, CronTool):
-            cron_token = cron_tool.set_cron_context(True)
-
-        message_record_token = None
-        if isinstance(message_tool, MessageTool):
-            message_record_token = message_tool.set_record_channel_delivery(True)
-
-        try:
-            resp = await agent.process_direct(
-                reminder_note,
-                session_key=f"cron:{job.id}",
-                channel=job.payload.channel or "cli",
-                chat_id=job.payload.to or "direct",
-                on_progress=_silent,
-            )
-        finally:
-            if isinstance(cron_tool, CronTool) and cron_token is not None:
-                cron_tool.reset_cron_context(cron_token)
-            if isinstance(message_tool, MessageTool) and message_record_token is not None:
-                message_tool.reset_record_channel_delivery(message_record_token)
-
-        response = resp.content if resp else ""
-
-        if job.payload.deliver and isinstance(message_tool, MessageTool) and message_tool._sent_in_turn:
-            return response
-
-        if job.payload.deliver and job.payload.to and response:
-            should_notify = await evaluate_response(
-                response, reminder_note, agent.provider, agent.model,
-            )
-            if should_notify:
-                await _deliver_to_channel(
-                    OutboundMessage(
-                        channel=job.payload.channel or "cli",
-                        chat_id=job.payload.to,
-                        content=response,
-                        metadata=dict(job.payload.channel_meta),
-                    ),
-                    record=True,
-                    session_key=job.payload.session_key,
-                )
-        return response
-
-    cron.on_job = on_cron_job
+    cron.on_job = cron_executor.run
 
     def _webui_runtime_model_name() -> str | None:
         model = getattr(agent, "model", None)
@@ -1164,20 +1033,6 @@ def _run_gateway(
         webui_runtime_capabilities=webui_runtime_capabilities,
     )
 
-    def _pick_heartbeat_target() -> tuple[str, str]:
-        """Pick a routable channel/chat target for heartbeat-triggered messages."""
-        enabled = set(channels.enabled_channels)
-        for item in session_manager.list_sessions():
-            key = item.get("key") or ""
-            if ":" not in key:
-                continue
-            channel, chat_id = key.split(":", 1)
-            if channel in {"cli", "system"}:
-                continue
-            if channel in enabled and chat_id:
-                return channel, chat_id
-        return "cli", "direct"
-
     if channels.enabled_channels:
         console.print(f"[green]✓[/green] Channels enabled: {', '.join(channels.enabled_channels)}")
     else:
@@ -1187,7 +1042,6 @@ def _run_gateway(
     if cron_status["jobs"] > 0:
         console.print(f"[green]✓[/green] Cron: {cron_status['jobs']} scheduled jobs")
 
-    hb_cfg = config.gateway.heartbeat
     if hb_cfg.enabled:
         console.print(f"[green]✓[/green] Heartbeat: every {hb_cfg.interval_s}s")
     else:

nanobot/cron/executor.py

@@ -0,0 +1,354 @@
+"""Cron job execution for the gateway runtime."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Awaitable, Callable
+from pathlib import Path
+from typing import Any, Protocol
+
+from loguru import logger
+
+import nanobot.utils.evaluator as evaluator
+from nanobot.agent.tools.cron import CronTool
+from nanobot.agent.tools.message import MessageTool
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.cron.types import CronJob
+
+
+class DeliverToChannel(Protocol):
+    def __call__(
+        self,
+        msg: OutboundMessage,
+        *,
+        record: bool = False,
+        session_key: str | None = None,
+    ) -> Awaitable[None]: ...
+
+
+ChannelLookup = Callable[[str], Any | None]
+EvaluateResponse = Callable[..., Awaitable[bool]]
+HeartbeatTaskDetector = Callable[[str], bool]
+HeartbeatTargetPicker = Callable[[], tuple[str, str]]
+
+
+class _CronStreamBuffer:
+    def __init__(
+        self,
+        *,
+        channel: str,
+        chat_id: str,
+        channel_meta: dict[str, Any],
+        base_id: str,
+    ) -> None:
+        self.channel = channel
+        self.chat_id = chat_id
+        self.channel_meta = channel_meta
+        self.base_id = base_id
+        self.segment = 0
+        self.events: list[OutboundMessage] = []
+        self.has_delta = False
+
+    def _stream_id(self) -> str:
+        return f"{self.base_id}:{self.segment}"
+
+    async def on_stream(self, delta: str) -> None:
+        meta = dict(self.channel_meta)
+        meta["_stream_delta"] = True
+        meta["_stream_id"] = self._stream_id()
+        self.events.append(OutboundMessage(
+            channel=self.channel,
+            chat_id=self.chat_id,
+            content=delta,
+            metadata=meta,
+        ))
+        if delta:
+            self.has_delta = True
+
+    async def on_stream_end(self, *, resuming: bool = False) -> None:
+        meta = dict(self.channel_meta)
+        meta["_stream_end"] = True
+        meta["_resuming"] = resuming
+        meta["_stream_id"] = self._stream_id()
+        self.events.append(OutboundMessage(
+            channel=self.channel,
+            chat_id=self.chat_id,
+            content="",
+            metadata=meta,
+        ))
+        self.segment += 1
+
+    async def publish(self, bus: MessageBus) -> None:
+        for event in self.events:
+            await bus.publish_outbound(event)
+
+
+class CronJobExecutor:
+    """Runs scheduled cron jobs through the agent and optional channel delivery."""
+
+    def __init__(
+        self,
+        *,
+        agent: Any,
+        bus: MessageBus,
+        deliver_to_channel: DeliverToChannel,
+        get_channel: ChannelLookup | None = None,
+        evaluate_response: EvaluateResponse | None = None,
+        heartbeat_workspace: Path | None = None,
+        heartbeat_preamble: str = "",
+        heartbeat_has_active_tasks: HeartbeatTaskDetector | None = None,
+        pick_heartbeat_target: HeartbeatTargetPicker | None = None,
+        heartbeat_keep_recent_messages: int = 8,
+    ) -> None:
+        self.agent = agent
+        self.bus = bus
+        self.deliver_to_channel = deliver_to_channel
+        self.get_channel = get_channel or (lambda _channel: None)
+        self.evaluate_response = evaluate_response or evaluator.evaluate_response
+        self.heartbeat_workspace = heartbeat_workspace
+        self.heartbeat_preamble = heartbeat_preamble
+        self.heartbeat_has_active_tasks = heartbeat_has_active_tasks
+        self.pick_heartbeat_target = pick_heartbeat_target
+        self.heartbeat_keep_recent_messages = heartbeat_keep_recent_messages
+
+    async def run(self, job: CronJob) -> str | None:
+        if job.name == "dream":
+            return await self._run_dream()
+        if job.name == "heartbeat":
+            return await self._run_heartbeat()
+
+        return await self._run_agent_turn(job)
+
+    async def _run_dream(self) -> None:
+        from nanobot.agent.memory import MemoryStore
+
+        dream_session_key = MemoryStore.dream_session_key
+        build_dream_commit_message = MemoryStore.build_dream_commit_message
+        prune_dream_sessions = MemoryStore.prune_dream_sessions
+
+        store = self.agent.context.memory
+        resp = None
+        try:
+            result = store.build_dream_prompt()
+            if result is None:
+                logger.info("Dream: nothing to process")
+                return None
+            prompt, last_cursor = result
+            resp = await self.agent.process_direct(
+                prompt,
+                session_key=dream_session_key(),
+                ephemeral=True,
+                tools=store.build_dream_tools(),
+                on_progress=self._silent,
+            )
+            if MemoryStore.dream_run_completed(resp):
+                store.set_last_dream_cursor(last_cursor)
+                logger.info("Dream cron job completed, cursor advanced to {}", last_cursor)
+            else:
+                logger.warning(
+                    "Dream cron job did not complete; cursor remains at {}",
+                    store.get_last_dream_cursor(),
+                )
+        except Exception:
+            logger.exception("Dream cron job failed")
+        finally:
+            if store.git.is_initialized():
+                msg = build_dream_commit_message(
+                    "dream: periodic memory consolidation", resp,
+                )
+                sha = store.git.auto_commit(msg)
+                if sha:
+                    logger.info("Dream commit: {}", sha)
+            store.compact_history()
+            prune_dream_sessions(self.agent.sessions.sessions_dir)
+        return None
+
+    async def _run_heartbeat(self) -> str | None:
+        if (
+            self.heartbeat_workspace is None
+            or self.heartbeat_has_active_tasks is None
+            or self.pick_heartbeat_target is None
+        ):
+            logger.warning("Heartbeat cron job skipped: executor is not configured for heartbeat")
+            return None
+
+        heartbeat_file = self.heartbeat_workspace / "HEARTBEAT.md"
+        try:
+            content = heartbeat_file.read_text(encoding="utf-8")
+        except OSError:
+            logger.debug("Heartbeat: HEARTBEAT.md missing")
+            return None
+        if not self.heartbeat_has_active_tasks(content):
+            logger.debug("Heartbeat: HEARTBEAT.md has no active tasks")
+            return None
+
+        channel, chat_id = self.pick_heartbeat_target()
+        if channel == "cli":
+            return None
+
+        prompt = (
+            self.heartbeat_preamble
+            + f"Review the following HEARTBEAT.md and report any active tasks:\n\n{content}"
+        )
+
+        message_tool = self._tool("message")
+        suppress_token = None
+        if isinstance(message_tool, MessageTool):
+            suppress_token = message_tool.set_suppress_delivery(True)
+        try:
+            resp = await self.agent.process_direct(
+                prompt,
+                session_key="heartbeat",
+                channel=channel,
+                chat_id=chat_id,
+                on_progress=self._silent,
+            )
+        finally:
+            if isinstance(message_tool, MessageTool) and suppress_token is not None:
+                message_tool.reset_suppress_delivery(suppress_token)
+        response = resp.content if resp else ""
+
+        session = self.agent.sessions.get_or_create("heartbeat")
+        session.retain_recent_legal_suffix(self.heartbeat_keep_recent_messages)
+        self.agent.sessions.save(session)
+
+        if not response:
+            return None
+
+        should_notify = await self.evaluate_response(
+            response, prompt, self.agent.provider, self.agent.model,
+            default_notify=False,
+        )
+        if should_notify:
+            logger.info("Heartbeat: completed, delivering response")
+            await self.deliver_to_channel(
+                OutboundMessage(channel=channel, chat_id=chat_id, content=response),
+                record=True,
+            )
+        else:
+            logger.info("Heartbeat: silenced by post-run evaluation")
+        return response
+
+    async def _run_agent_turn(self, job: CronJob) -> str | None:
+        reminder_note = self._reminder_note(job)
+        cron_tool = self._tool("cron")
+        cron_token = None
+        if isinstance(cron_tool, CronTool):
+            cron_token = cron_tool.set_cron_context(True)
+
+        message_tool = self._tool("message")
+        message_record_token = None
+        if isinstance(message_tool, MessageTool):
+            message_record_token = message_tool.set_record_channel_delivery(True)
+
+        channel_name = job.payload.channel or "cli"
+        chat_id = job.payload.to or "direct"
+        stream = self._stream_buffer(job, channel_name=channel_name, chat_id=chat_id)
+
+        try:
+            resp = await self.agent.process_direct(
+                reminder_note,
+                session_key=f"cron:{job.id}",
+                channel=channel_name,
+                chat_id=chat_id,
+                on_progress=self._silent,
+                on_stream=stream.on_stream if stream else None,
+                on_stream_end=stream.on_stream_end if stream else None,
+            )
+        finally:
+            if isinstance(cron_tool, CronTool) and cron_token is not None:
+                cron_tool.reset_cron_context(cron_token)
+            if isinstance(message_tool, MessageTool) and message_record_token is not None:
+                message_tool.reset_record_channel_delivery(message_record_token)
+
+        response = resp.content if resp else ""
+
+        if job.payload.deliver and isinstance(message_tool, MessageTool) and message_tool._sent_in_turn:
+            await self._publish_turn_end_if_needed(job, channel_name=channel_name, chat_id=chat_id)
+            return response
+
+        delivered = False
+        if job.payload.deliver and job.payload.to and response:
+            should_notify = await self.evaluate_response(
+                response, reminder_note, self.agent.provider, self.agent.model,
+            )
+            if should_notify:
+                meta = dict(job.payload.channel_meta)
+                if stream and stream.has_delta:
+                    await stream.publish(self.bus)
+                    meta["_streamed"] = True
+                await self.deliver_to_channel(
+                    OutboundMessage(
+                        channel=channel_name,
+                        chat_id=chat_id,
+                        content=response,
+                        metadata=meta,
+                    ),
+                    record=True,
+                    session_key=job.payload.session_key,
+                )
+                delivered = True
+
+        if delivered:
+            await self._publish_turn_end_if_needed(job, channel_name=channel_name, chat_id=chat_id)
+        return response
+
+    def _tool(self, name: str) -> Any | None:
+        tools = getattr(self.agent, "tools", {})
+        if hasattr(tools, "get"):
+            return tools.get(name)
+        return None
+
+    def _stream_buffer(
+        self,
+        job: CronJob,
+        *,
+        channel_name: str,
+        chat_id: str,
+    ) -> _CronStreamBuffer | None:
+        target_channel = self.get_channel(channel_name)
+        wants_stream = bool(
+            job.payload.deliver
+            and job.payload.to
+            and target_channel is not None
+            and target_channel.supports_streaming
+        )
+        if not wants_stream:
+            return None
+        return _CronStreamBuffer(
+            channel=channel_name,
+            chat_id=chat_id,
+            channel_meta=job.payload.channel_meta,
+            base_id=f"cron:{job.id}:{time.time_ns()}",
+        )
+
+    async def _publish_turn_end_if_needed(
+        self,
+        job: CronJob,
+        *,
+        channel_name: str,
+        chat_id: str,
+    ) -> None:
+        if channel_name != "websocket" or not job.payload.to:
+            return
+        await self.bus.publish_outbound(OutboundMessage(
+            channel=channel_name,
+            chat_id=chat_id,
+            content="",
+            metadata={**job.payload.channel_meta, "_turn_end": True},
+        ))
+
+    @staticmethod
+    async def _silent(*_args: Any, **_kwargs: Any) -> None:
+        pass
+
+    @staticmethod
+    def _reminder_note(job: CronJob) -> str:
+        return (
+            "The scheduled time has arrived. Deliver this reminder to the user now, "
+            "as a brief and natural message in their language. Speak directly to them — "
+            "do not narrate progress, summarize, include user IDs, or add status reports "
+            "like 'Done' or 'Reminded'.\n\n"
+            f"Reminder: {job.payload.message}"
+        )

tests/cli/test_commands.py

@@ -1421,6 +1421,262 @@ def test_gateway_cron_job_suppresses_intermediate_progress(
     bus.publish_outbound.assert_not_awaited()
 
 
+def test_gateway_cron_job_streams_when_channel_supports_it(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """Cron jobs on streaming channels must emit deltas with stream_id and turn_end."""
+    config_file = tmp_path / "instance" / "config.json"
+    config_file.parent.mkdir(parents=True)
+    config_file.write_text("{}")
+
+    config = Config()
+    config.agents.defaults.workspace = str(tmp_path / "config-workspace")
+    bus = MagicMock()
+    bus.publish_outbound = AsyncMock()
+    seen: dict[str, object] = {}
+
+    monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
+    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
+    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
+    monkeypatch.setattr(
+        "nanobot.providers.factory.build_provider_snapshot",
+        lambda _config: _test_provider_snapshot(object(), _config),
+    )
+    monkeypatch.setattr(
+        "nanobot.providers.factory.load_provider_snapshot",
+        lambda _config_path=None: _test_provider_snapshot(object(), config),
+    )
+    monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: bus)
+    monkeypatch.setattr("nanobot.session.manager.SessionManager", lambda _workspace: object())
+
+    async def _always_notify(*_args, **_kwargs) -> bool:
+        return True
+
+    class _FakeStreamingChannel:
+        supports_streaming = True
+
+    class _FakeChannelManager:
+        def __init__(self, *_args, **_kwargs) -> None:
+            self.channels = {"websocket": _FakeStreamingChannel()}
+            self.enabled_channels = ["websocket"]
+
+        async def start_all(self):
+            pass
+
+        async def stop_all(self):
+            pass
+
+    class _FakeCron:
+        def __init__(self, _store_path: Path) -> None:
+            self.on_job = None
+            seen["cron"] = self
+
+        def status(self):
+            return {"enabled": True, "jobs": 0, "next_wake_at_ms": None}
+
+        def register_system_job(self, job):
+            pass
+
+        def stop(self):
+            pass
+
+    class _FakeAgentLoop:
+        @classmethod
+        def from_config(cls, config, bus=None, **extra):
+            return cls(**extra)
+        def __init__(self, *args, **kwargs) -> None:
+            self.model = "test-model"
+            self.provider = object()
+            self.tools = {}
+            self.dream = MagicMock()
+            self.sessions = MagicMock()
+
+        async def process_direct(self, *_args, on_stream=None, on_stream_end=None, **_kwargs):
+            seen["on_stream"] = on_stream
+            seen["on_stream_end"] = on_stream_end
+            if on_stream:
+                await on_stream("Hello")
+                await on_stream(" world")
+            if on_stream_end:
+                await on_stream_end(resuming=False)
+            return OutboundMessage(
+                channel="websocket",
+                chat_id="user-1",
+                content="Hello world",
+            )
+
+        async def close_mcp(self) -> None:
+            return None
+
+        async def run(self) -> None:
+            return None
+
+        def stop(self) -> None:
+            return None
+
+    monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
+    monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
+    monkeypatch.setattr(
+        "nanobot.cli.commands.evaluate_response",
+        _always_notify,
+    )
+
+    result = runner.invoke(app, ["gateway", "--config", str(config_file)])
+    assert result.exit_code == 0
+
+    cron = seen["cron"]
+    job = CronJob(
+        id="cron-stream-test",
+        name="test-stream",
+        payload=CronPayload(
+            message="Say hello.",
+            deliver=True,
+            channel="websocket",
+            to="user-1",
+        ),
+    )
+    response = asyncio.run(cron.on_job(job))
+
+    assert response == "Hello world"
+    assert seen["on_stream"] is not None
+    assert seen["on_stream_end"] is not None
+
+    calls = bus.publish_outbound.await_args_list
+    # First two calls are streaming deltas
+    assert calls[0].args[0].metadata.get("_stream_delta") is True
+    assert calls[0].args[0].metadata.get("_stream_id") is not None
+    assert calls[0].args[0].content == "Hello"
+    assert calls[1].args[0].metadata.get("_stream_delta") is True
+    assert calls[1].args[0].metadata.get("_stream_id") == calls[0].args[0].metadata["_stream_id"]
+    assert calls[1].args[0].content == " world"
+    # Third call is stream_end
+    assert calls[2].args[0].metadata.get("_stream_end") is True
+    assert calls[2].args[0].metadata.get("_stream_id") == calls[0].args[0].metadata["_stream_id"]
+    # Fourth call is the final message with _streamed marker
+    assert calls[3].args[0].metadata.get("_streamed") is True
+    assert calls[3].args[0].content == "Hello world"
+    # Fifth call is turn_end
+    assert calls[4].args[0].metadata.get("_turn_end") is True
+
+
+def test_gateway_cron_job_streaming_respects_disabled_delivery(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """Streaming cron output must not reach the channel when delivery is disabled."""
+    config_file = tmp_path / "instance" / "config.json"
+    config_file.parent.mkdir(parents=True)
+    config_file.write_text("{}")
+
+    config = Config()
+    config.agents.defaults.workspace = str(tmp_path / "config-workspace")
+    bus = MagicMock()
+    bus.publish_outbound = AsyncMock()
+    seen: dict[str, object] = {}
+
+    monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
+    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
+    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
+    monkeypatch.setattr(
+        "nanobot.providers.factory.build_provider_snapshot",
+        lambda _config: _test_provider_snapshot(object(), _config),
+    )
+    monkeypatch.setattr(
+        "nanobot.providers.factory.load_provider_snapshot",
+        lambda _config_path=None: _test_provider_snapshot(object(), config),
+    )
+    monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: bus)
+    monkeypatch.setattr("nanobot.session.manager.SessionManager", lambda _workspace: object())
+
+    class _FakeStreamingChannel:
+        supports_streaming = True
+
+    class _FakeChannelManager:
+        def __init__(self, *_args, **_kwargs) -> None:
+            self.channels = {"websocket": _FakeStreamingChannel()}
+            self.enabled_channels = ["websocket"]
+
+        async def start_all(self):
+            pass
+
+        async def stop_all(self):
+            pass
+
+    class _FakeCron:
+        def __init__(self, _store_path: Path) -> None:
+            self.on_job = None
+            seen["cron"] = self
+
+        def status(self):
+            return {"enabled": True, "jobs": 0, "next_wake_at_ms": None}
+
+        def register_system_job(self, job):
+            pass
+
+        def stop(self):
+            pass
+
+    class _FakeAgentLoop:
+        @classmethod
+        def from_config(cls, config, bus=None, **extra):
+            return cls(**extra)
+        def __init__(self, *args, **kwargs) -> None:
+            self.model = "test-model"
+            self.provider = object()
+            self.tools = {}
+            self.dream = MagicMock()
+            self.sessions = MagicMock()
+
+        async def process_direct(self, *_args, on_stream=None, on_stream_end=None, **_kwargs):
+            seen["on_stream"] = on_stream
+            seen["on_stream_end"] = on_stream_end
+            if on_stream:
+                await on_stream("This should not leak")
+            if on_stream_end:
+                await on_stream_end(resuming=False)
+            return OutboundMessage(
+                channel="websocket",
+                chat_id="user-1",
+                content="This should not leak",
+            )
+
+        async def close_mcp(self) -> None:
+            return None
+
+        async def run(self) -> None:
+            return None
+
+        def stop(self) -> None:
+            return None
+
+    monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
+    monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
+
+    result = runner.invoke(app, ["gateway", "--config", str(config_file)])
+    assert result.exit_code == 0
+
+    cron = seen["cron"]
+    job = CronJob(
+        id="cron-stream-rejected-test",
+        name="test-stream-rejected",
+        payload=CronPayload(
+            message="Say something optional.",
+            deliver=False,
+            channel="websocket",
+            to="user-1",
+        ),
+    )
+    response = asyncio.run(cron.on_job(job))
+
+    assert response == "This should not leak"
+    assert seen["on_stream"] is None
+    assert seen["on_stream_end"] is None
+    bus.publish_outbound.assert_not_awaited()
+
+
 def test_gateway_workspace_override_does_not_migrate_legacy_cron(
     monkeypatch, tmp_path: Path
 ) -> None: