M5: push (FCM + ntfy) + offline catch-up + background notifications

Server (gateway-plugin):
- push.py: FCM (HTTP v1 service-account / legacy key) + ntfy backends; NtfyBackend.server_url for app discovery
- adapter.py: push on offline broadcast + high-priority push when live; fcm.register; server_caps.push_ntfy_server; outbox now ALWAYS appends so a reconnecting app catches up on live-delivered frames (fixes empty chat after notification tap / activity recreation)
- protocol.py / ws_server.py / outbox.py / plugin.yaml: M5 frames + env vars

App (Kotlin CMP):
- Protocol.kt: notification / fcm.register / sync frames + push caps
- GatewayClient.kt: hello carries push creds; auto-sync on reconnect
- IrisController.kt: banners, deep-link, notifyMessageIfBackgrounded (system notification on a regular reply when backgrounded)
- Android: PlatformPush, AppBridge, IrisNotifications, NtfyListenerService, IrisFirebaseMessagingService, AndroidPush, AndroidSecureStore
- Desktop: DesktopPush, DesktopSecureStore
- build files + manifest (permissions, services, deep-links)

Tests: 35-test tests/gateway/test_android.py suite passes (incl. new regression test_live_delivered_frame_still_parked_for_sync). E2E verified on device: push fire, reconnect sync, background notification, and message replay after ChatStore reset.

gateway-plugin/adapter.py

@@ -30,6 +30,14 @@ register the (delivery-validated) file in the media registry and emit
 ``media.offer``; ``media.pull`` streams the file back as chunked binary
 frames, re-checking ``validate_media_delivery_path`` at pull time.
 
+Milestone M5: push + offline. Frames with no live subscriber are parked in
+the outbox (M3) AND wake the device via the push backend (``push.py``: FCM
+HTTP v1 primary, ntfy fallback, selected by ``ANDROID_PUSH_BACKEND``).
+``notification`` frames render in-app banners and mirror to push (channel
+events, cron deliveries, approvals, clarifies); high-priority kinds push even
+when a device is live. ``fcm.register`` rotates push tokens (registry + live
+connection). The outbox enforces a row cap with a throttled prune notice.
+
 Configuration in config.yaml::
 
     gateway:
@@ -106,6 +114,7 @@ from . import protocol  # noqa: E402
 from . import search as search_bridge  # noqa: E402
 from .channels import get_directory  # noqa: E402
 from .outbox import Outbox  # noqa: E402
+from .push import NtfyBackend, PushBackend, build_push_backend  # noqa: E402
 from .pairing import (  # noqa: E402
     DeviceRegistry,
     generate_token,
@@ -280,6 +289,42 @@ def _strip_streaming_cursor(text: str) -> str:
     return text
 
 
+def _push_preview(text: Any, limit: int = 120) -> str:
+    """Short single-line preview for push bodies (lock-screen privacy: no
+    secrets, no full bodies -- full content arrives via ``sync``)."""
+    s = " ".join(str(text or "").split())
+    if len(s) > limit:
+        s = s[: limit - 1] + "…"
+    return s
+
+
+# Cron delivery wrap (cron/scheduler.py ``_deliver_result``,
+# cron.wrap_response: true):
+#   "Cronjob Response: <name>\n(job_id: <id>)\n-------------\n\n<content>\n\n
+#    To stop or manage this job, send me a new message (e.g. ...)."
+_CRON_WRAP_RE = re.compile(
+    r"^Cronjob Response: (.+?)\n\(job_id: [^)]*\)\n-+\n\n"
+)
+_CRON_FOOTER = "\n\nTo stop or manage this job"
+
+
+def _cron_brief(content: str, job_id: str) -> Tuple[str, str]:
+    """Parse a cron delivery into ``(job_name, inner_text)``.
+
+    Falls back to ``(job_id, content)`` when the wrap is disabled
+    (``cron.wrap_response: false``) or unrecognised.
+    """
+    m = _CRON_WRAP_RE.match(content or "")
+    if not m:
+        return str(job_id or "cron"), (content or "").strip()
+    name = m.group(1).strip()
+    body = content[m.end():]
+    idx = body.rfind(_CRON_FOOTER)
+    if idx != -1:
+        body = body[:idx]
+    return name, body.strip()
+
+
 def _split_reasoning(text: str) -> Tuple[Optional[str], str]:
     """Split a code-style reasoning prefix off the front of *text*.
 
@@ -693,6 +738,17 @@ class AndroidAdapter(BasePlatformAdapter):
         # last finalized assistant message id per chat (offer association).
         self._media = media_bridge.MediaStore(get_hermes_home())
         self._last_message_id: Dict[str, str] = {}
+        # M5: push backend (FCM primary, ntfy fallback) + the throttle for
+        # the outbox-prune banner.
+        self._push: PushBackend = build_push_backend(
+            self.push_backend,
+            fcm_service_account=_get_scoped_secret("ANDROID_FCM_SERVICE_ACCOUNT"),
+            fcm_server_key=_get_scoped_secret("ANDROID_FCM_SERVER_KEY"),
+            ntfy_topic=_get_scoped_secret("NTFY_TOPIC"),
+            ntfy_server_url=os.getenv("NTFY_SERVER_URL", "").strip() or None,
+            ntfy_auth_token=_get_scoped_secret("NTFY_AUTH_TOKEN"),
+        )
+        self._prune_notified_at = 0.0
 
     def _turn_state(self, chat_id: str) -> _TurnState:
         st = self._turns.get(chat_id)
@@ -747,6 +803,16 @@ class AndroidAdapter(BasePlatformAdapter):
         except Exception:
             logger.warning("android: ensure_default failed", exc_info=True)
 
+        # M5: push backend status (degrade gracefully when unconfigured).
+        if not self._push.configured():
+            logger.warning(
+                "android: push backend %r not configured (no credentials) -- "
+                "offline devices will not be woken; outbox + sync still apply",
+                self.push_backend,
+            )
+        else:
+            logger.info("android: push backend: %s", self._push.name)
+
         self._connected = True
         self._mark_connected()
         logger.info("android: connected; WS server on %s:%s", self.host, self.port)
@@ -817,7 +883,45 @@ class AndroidAdapter(BasePlatformAdapter):
             )
             return SendResult(success=True, message_id=message_id)
 
-        # 2. Final message (non-streaming final, or streaming fallback final).
+        # 2. Cron delivery (the cron scheduler passes ``job_id`` in metadata):
+        #    a final assistant message + a high-priority notification banner
+        #    (pushed even when a device is live, docs/08 §8.1).
+        if meta.get("job_id"):
+            reasoning, _ = _split_reasoning(content)
+            if reasoning:
+                _reset_reasoning()
+            else:
+                await _wait_for_reasoning_flushed()
+                reasoning = _take_reasoning() or None
+            name, inner = _cron_brief(content, str(meta.get("job_id")))
+            message_id = _mint_message_id()
+            await self._broadcast_or_log(
+                chat_id,
+                protocol.notification(
+                    chat_id,
+                    protocol.NOTIF_CRON,
+                    f"Cron: {name}",
+                    _push_preview(inner),
+                    thread_id=thread_id,
+                ),
+            )
+            await self._broadcast_or_log(
+                chat_id,
+                protocol.message(
+                    chat_id=chat_id,
+                    message_id=message_id,
+                    role=protocol.ROLE_ASSISTANT,
+                    text=inner,
+                    thread_id=thread_id,
+                    reasoning=reasoning,
+                    ts=int(time.time() * 1000),
+                ),
+            )
+            self._last_message_id[chat_id] = message_id
+            state.active = False
+            return SendResult(success=True, message_id=message_id)
+
+        # 3. Final message (non-streaming final, or streaming fallback final).
         if meta.get("notify") is True:
             reasoning, body = _split_reasoning(content)
             # Non-streaming: reasoning is prepended to content (split above).
@@ -863,11 +967,11 @@ class AndroidAdapter(BasePlatformAdapter):
             state.active = False
             return SendResult(success=True, message_id=message_id)
 
-        # 3. Tool progress (first tool bubble of an editable line buffer).
+        # 4. Tool progress (first tool bubble of an editable line buffer).
         if _is_tool_progress(content):
             return await self._emit_tool_lines(chat_id, content, state, thread_id, is_edit=False)
 
-        # 4. Commentary (interim assistant beat).
+        # 5. Commentary (interim assistant beat).
         message_id = _mint_message_id()
         state.active = True
         await self._broadcast_or_log(
@@ -1043,17 +1147,151 @@ class AndroidAdapter(BasePlatformAdapter):
 
     async def _broadcast_or_log(self, chat_id: str, frame: "protocol.Frame") -> None:
         delivered = await self._ws_server.broadcast(frame)
+        # M3/M5: always append to the outbox so a reconnecting app can catch
+        # up on *all* recent frames, not just the ones that were parked. This
+        # covers the case where the app's in-memory ChatStore is reset (e.g.
+        # activity recreation / composition recompose) while the process stays
+        # alive: the recreated controller re-syncs from its stale cursor and
+        # replays the frames it missed. The push below still only fires when
+        # there is no live subscriber (or for high-priority events).
+        try:
+            cursor = self._outbox.append(chat_id, frame.to_json())
+        except Exception:
+            logger.warning("android: outbox append failed", exc_info=True)
+            return
         if delivered == 0:
-            # M3: no live device -- park the frame in the outbox so a
-            # reconnecting app can `sync` it (M5 adds push to wake the device).
+            logger.info(
+                "android: no live devices for %s; %s frame parked in outbox (cursor=%s)",
+                chat_id, frame.type, cursor,
+            )
+            # M5: wake the offline device(s) via the push backend.
+            await self._maybe_push(chat_id, frame, cursor)
+            await self._maybe_notify_outbox_prune(chat_id)
+        elif (
+            frame.type == protocol.TYPE_NOTIFICATION
+            and frame.payload.get("kind") in protocol.HIGH_PRIORITY_NOTIF_KINDS
+        ):
+            # M5: high-priority events (approval/clarify/cron) push even when
+            # a device is live -- the app may be backgrounded and decides
+            # whether to also show an in-app banner (docs/08 §8.1).
+            await self._maybe_push(chat_id, frame, cursor)
+
+    # ── M5: push ───────────────────────────────────────────────────────────
+
+    def _push_summary(
+        self, frame: "protocol.Frame"
+    ) -> Optional[Tuple[str, str, str, str]]:
+        """``(title, body, kind, priority)`` for a pushable frame, else None.
+
+        Only terminal/interesting frames wake a device: intermediate
+        streaming and tool frames are replayed by ``sync`` without a push
+        (no notification spam per turn).
+        """
+        t = frame.type
+        p = frame.payload
+        if t == protocol.TYPE_MESSAGE:
+            return (
+                self._channel_name(frame.chat_id or ""),
+                _push_preview(p.get("text")),
+                "message",
+                "normal",
+            )
+        if t == protocol.TYPE_MESSAGE_STOP:
+            return (
+                self._channel_name(frame.chat_id or ""),
+                _push_preview(p.get("final_text")),
+                "message",
+                "normal",
+            )
+        if t == protocol.TYPE_NOTIFICATION:
+            kind = str(p.get("kind") or protocol.NOTIF_GENERIC)
+            priority = "high" if kind in protocol.HIGH_PRIORITY_NOTIF_KINDS else "normal"
+            return (
+                str(p.get("title") or "Iris"),
+                str(p.get("body") or ""),
+                kind,
+                priority,
+            )
+        if t == protocol.TYPE_MEDIA_OFFER:
+            return (
+                self._channel_name(frame.chat_id or ""),
+                f"New {p.get('kind') or 'media'}: {p.get('filename') or ''}".strip(),
+                "media",
+                "normal",
+            )
+        return None
+
+    async def _maybe_push(self, chat_id: str, frame: "protocol.Frame", cursor: int) -> None:
+        """Fire the configured push backend for a parked (or high-priority)
+        frame. Best-effort: failures are logged, never raised."""
+        summary = self._push_summary(frame)
+        if summary is None:
+            return
+        title, body, kind, priority = summary
+        backend = self._push
+        if backend is None or not backend.token_field:
+            return
+        devices = self._devices.list()
+        if not backend.configured() and not any(
+            d.get(backend.token_field) for d in devices
+        ):
+            return
+        data: Dict[str, Any] = {"chat_id": chat_id, "kind": kind, "cursor": str(cursor)}
+        if frame.thread_id:
+            data["thread_id"] = frame.thread_id
+        message_id = frame.payload.get("message_id")
+        if isinstance(message_id, str) and message_id:
+            data["message_id"] = message_id
+        for device in devices:
+            device_id = device.get("device_id")
+            if not device_id:
+                continue
+            # Prefer the live connection's token (fcm.register refreshes it
+            # in memory) over the possibly-stale registry row.
+            conn = self._ws_server.connection(device_id)
+            token = getattr(conn, backend.token_field, None) if conn is not None else None
+            if not token:
+                token = device.get(backend.token_field)
+            if not token:
+                continue
             try:
-                cursor = self._outbox.append(chat_id, frame.to_json())
-                logger.info(
-                    "android: no live devices for %s; %s frame parked in outbox (cursor=%s)",
-                    chat_id, frame.type, cursor,
+                ok = await backend.send(
+                    device_id=device_id,
+                    chat_id=chat_id,
+                    title=title,
+                    body=body,
+                    data=data,
+                    token=token,
+                    priority=priority,
                 )
             except Exception:
-                logger.warning("android: outbox append failed", exc_info=True)
+                logger.warning("android: push via %s failed", backend.name, exc_info=True)
+                continue
+            if ok:
+                logger.info(
+                    "android: push via %s -> %s (%s, chat=%s)",
+                    backend.name, device_id, frame.type, chat_id,
+                )
+
+    async def _maybe_notify_outbox_prune(self, chat_id: str) -> None:
+        """When the outbox row cap pruned old frames, tell the app (throttled
+        to once per hour so a full box doesn't banner per frame)."""
+        pruned = self._outbox.take_overflow_pruned()
+        if pruned <= 0:
+            return
+        now = time.time()
+        if now - self._prune_notified_at < 3600.0:
+            return
+        self._prune_notified_at = now
+        await self._broadcast_or_log(
+            chat_id,
+            protocol.notification(
+                chat_id,
+                protocol.NOTIF_GENERIC,
+                "Outbox",
+                f"{pruned} older message(s) pruned",
+            ),
+        )
 
     async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None:
         """Send a typing indicator (``typing`` frame, on=true)."""
@@ -1464,6 +1702,16 @@ class AndroidAdapter(BasePlatformAdapter):
         resp = protocol.channel_created(entry)
         resp.id = frame.id
         await self._ws_server.broadcast(resp)
+        # M5: banner + push mirror (parked in the outbox when offline).
+        await self._broadcast_or_log(
+            entry["chat_id"],
+            protocol.notification(
+                entry["chat_id"],
+                protocol.NOTIF_CHANNEL_CREATED,
+                "Channels",
+                f"New channel: {name}",
+            ),
+        )
 
     async def on_channel_rename(self, frame: protocol.Frame, device_id: str) -> None:
         chat_id = frame.chat_id or frame.payload.get("chat_id")
@@ -1496,6 +1744,16 @@ class AndroidAdapter(BasePlatformAdapter):
         resp = protocol.channel_renamed(entry)
         resp.id = frame.id
         await self._ws_server.broadcast(resp)
+        # M5: banner + push mirror (parked in the outbox when offline).
+        await self._broadcast_or_log(
+            chat_id,
+            protocol.notification(
+                chat_id,
+                protocol.NOTIF_CHANNEL_RENAMED,
+                "Channels",
+                f"Renamed to {name}",
+            ),
+        )
 
     async def on_channel_set_default(self, frame: protocol.Frame, device_id: str) -> None:
         chat_id = frame.chat_id or frame.payload.get("chat_id")
@@ -1536,6 +1794,16 @@ class AndroidAdapter(BasePlatformAdapter):
         resp = protocol.channel_deleted(chat_id)
         resp.id = frame.id
         await self._ws_server.broadcast(resp)
+        # M5: banner + push mirror (parked in the outbox when offline).
+        await self._broadcast_or_log(
+            chat_id,
+            protocol.notification(
+                chat_id,
+                protocol.NOTIF_CHANNEL_DELETED,
+                "Channels",
+                f"{entry.get('name') or chat_id} deleted",
+            ),
+        )
 
     async def on_channel_list(self, frame: protocol.Frame, device_id: str) -> None:
         channels = self._channels.list(include_archived=False)
@@ -1596,6 +1864,140 @@ class AndroidAdapter(BasePlatformAdapter):
         done = protocol.sync_done(self._outbox.latest_cursor(), id=frame.id)
         await self._ws_server.send_to(device_id, done)
 
+    # ── M5: push token registration ───────────────────────────────────────
+
+    async def on_fcm_register(self, frame: protocol.Frame, device_id: str) -> None:
+        """Update the device's push tokens (FCM rotation / ntfy topic).
+
+        Persists to the device registry AND refreshes the live connection so
+        the next push targets the current token without a stale read.
+        """
+        fcm_token = frame.payload.get("fcm_token")
+        ntfy_topic = frame.payload.get("ntfy_topic")
+        fcm_token = fcm_token if isinstance(fcm_token, str) and fcm_token else None
+        ntfy_topic = ntfy_topic if isinstance(ntfy_topic, str) and ntfy_topic else None
+        if fcm_token is None and ntfy_topic is None:
+            return
+        try:
+            self._devices.update_push_tokens(
+                device_id, fcm_token=fcm_token, ntfy_topic=ntfy_topic
+            )
+        except Exception:
+            logger.warning("android: fcm.register update failed", exc_info=True)
+            return
+        conn = self._ws_server.connection(device_id)
+        if conn is not None:
+            if fcm_token is not None:
+                conn.fcm_token = fcm_token
+            if ntfy_topic is not None:
+                conn.ntfy_topic = ntfy_topic
+        logger.info("android: push tokens updated for %s", device_id)
+
+    # ── M5: approval / clarify banners ────────────────────────────────────
+
+    async def send_slash_confirm(
+        self,
+        chat_id: str,
+        title: str,
+        message: str,
+        session_key: str,
+        confirm_id: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Banner + push for a slash-command approval prompt.
+
+        The gateway's text fallback still renders the actionable prompt (the
+        app has no inline buttons yet); the notification is the push-visible
+        signal (high priority: pushed even when a device is live).
+        """
+        thread_id = _thread_id_from_metadata(metadata)
+        await self._broadcast_or_log(
+            chat_id,
+            protocol.notification(
+                chat_id,
+                protocol.NOTIF_APPROVAL,
+                title or "Approval needed",
+                _push_preview(message),
+                thread_id=thread_id,
+            ),
+        )
+        return await super().send_slash_confirm(
+            chat_id, title, message, session_key, confirm_id, metadata=metadata
+        )
+
+    async def send_clarify(
+        self,
+        chat_id: str,
+        question: str,
+        choices: Optional[list],
+        clarify_id: str,
+        session_key: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Banner + push for a clarify prompt.
+
+        Renders the prompt as a proper ``message`` frame (the base text
+        fallback would route through ``send()`` and be misclassified as
+        commentary/tool progress) and keeps the gateway's text intercept
+        working via ``mark_awaiting_text``.
+        """
+        thread_id = _thread_id_from_metadata(metadata)
+        await self._broadcast_or_log(
+            chat_id,
+            protocol.notification(
+                chat_id,
+                protocol.NOTIF_CLARIFY,
+                "Question",
+                _push_preview(question),
+                thread_id=thread_id,
+            ),
+        )
+        if choices:
+            # Multi-select clarifies register their flag on the pending entry;
+            # look it up by id (mirrors the base text fallback).
+            _is_multi = False
+            try:
+                from tools import clarify_gateway as _cg
+
+                with _cg._lock:
+                    _entry = _cg._entries.get(clarify_id)
+                _is_multi = bool(_entry and getattr(_entry, "multi_select", False))
+            except Exception:
+                _is_multi = False
+            lines = [f"❓ {question}", ""]
+            for i, choice in enumerate(choices, start=1):
+                lines.append(f"  {i}. {choice}")
+            lines.append("")
+            if _is_multi:
+                lines.append(
+                    "Multiple selections allowed — reply with the numbers "
+                    "separated by commas or spaces (e.g. \"1, 3\"), the option "
+                    "text, or your own answer."
+                )
+            else:
+                lines.append("Reply with the number, the option text, or your own answer.")
+            text = "\n".join(lines)
+            # Text fallback: enable text-capture so the gateway intercept
+            # picks up the user's typed reply (e.g. "2" or choice text).
+            from tools.clarify_gateway import mark_awaiting_text
+
+            mark_awaiting_text(clarify_id)
+        else:
+            text = f"❓ {question}"
+        message_id = _mint_message_id()
+        await self._broadcast_or_log(
+            chat_id,
+            protocol.message(
+                chat_id=chat_id,
+                message_id=message_id,
+                role=protocol.ROLE_ASSISTANT,
+                text=text,
+                thread_id=thread_id,
+                ts=int(time.time() * 1000),
+            ),
+        )
+        return SendResult(success=True, message_id=message_id)
+
     # ── Chat info ─────────────────────────────────────────────────────────
 
     def _channel_name(self, chat_id: str) -> str:
@@ -1630,6 +2032,12 @@ class AndroidAdapter(BasePlatformAdapter):
             "media": True,       # M4: media.upload/offer/pull
             "search": True,      # M3: search frame
             "push": self.push_backend,
+            # M5: ntfy server URL (app listener discovery; "" when not ntfy).
+            "push_ntfy_server": (
+                self._push.server_url
+                if isinstance(self._push, NtfyBackend)
+                else ""
+            ),
             "pickers": False,    # M2+
         }