feat: Telegram messages get real formatting + push DMs at draft origination (#568)

* feat(telegram): HTML-styled bot messages + push DMs at held-draft origination

* fix(telegram): attr-context escaping, balance-aware truncation, send observability; docs

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

CLAUDE.md

@@ -414,7 +414,7 @@ Agent backends are pluggable. `roboco/llm/providers/` defines an `AgentProvider`
 
 **Env-branches ladder + EnvSyncEngine (default-off `ROBOCO_ENV_SYNC_ENABLED`).** Replaces a project's single `default_branch` with an ordered environment ladder: nullable `projects.environments` JSONB (migration 073), an ordered `list[{name, branch}]` where index 0 is the **head** rung (where dev/cell/leaf PRs land) and index -1 is the **prod** rung (where the gated release executor commits + tags); middle rungs are intermediates (qa/stag). A null ladder degenerates to a single-branch ladder synthesized from `default_branch` at read time (`roboco/models/env_branches.py`: `head_branch` / `prod_branch` / `ladder_pairs` / `promotion_chain`) — no backfill, byte-for-byte legacy behavior until the CEO declares a real split. Every former `default_branch` consumer now routes through the shim: the PR target and per-agent clone (`WorkspaceService.ensure_workspace` / `ensure_read_clone`), the CI branch, the release executor's clone/commit/tag target (`_ReleaseContext.prod_branch`) plus its full-chain head→…→prod promotion before bumping (`promote_env_chain`, fail-closed `promotion_failed` on a merge conflict), and `release_readiness`'s diff baseline (`prod..head` instead of `last_tag..HEAD`) with a tag-drift cross-check (`_tag_drift_gaps` — the last tag's commit vs. prod tip disagreeing flags a hotfix that landed on prod after the tag). `EnvSyncEngine` (`roboco/services/env_sync_engine.py`) cascades the ladder prod→…→head via GitHub's merges API: a clean merge auto-pushes straight to the lower rung, a conflict opens ONE idempotent sync PR + a Main-PM coordination task and stops that project's cascade for the cycle — the cascade's target is never the prod rung by construction, so "only the CEO merges master" still holds. Bounded + deduped per repo (one open env_sync task at a time). Panel: an environment-ladder editor on the project edit dialog.
 
-**Telegram notifications bridge V1+V2+V3 (default-off `ROBOCO_TELEGRAM_ENABLED`).** V1: best-effort, outbound-only Telegram DMs to the CEO on escalation and completion. Mirrors the `x_credentials` pattern: a singleton Fernet-encrypted `telegram_credentials` row (migration 074, bot token + chat id; the API returns `has_credentials` only) behind CEO-only `/telegram/credentials` routes and a panel credentials card. `_notify_telegram` (`roboco/services/notification_delivery.py`) fans out from `notify_ceo_of_escalation` / `notify_ceo_of_completion`, sending only the notification's subject plus an optional panel deep-link (`panel_base_url`) — never the body — via a deferred, best-effort send that never raises into the producer (`NullTelegramClient` when unconfigured or the flag is off, `LiveTelegramClient` posting to the Bot API otherwise). V2 (`ROBOCO_TELEGRAM_INBOUND_ENABLED`, sub-switch on top of V1's flag — both plus stored credentials are required, otherwise the bot only sends and never listens) makes the bridge two-way: `TelegramInboundEngine` (`roboco/services/telegram_inbound.py`) long-polls `getUpdates` from a dedicated orchestrator loop (`_telegram_poll_loop`), authorizing every update by BOTH chat id and sender id, and routes `/status` / `/queue` / `/task` commands plus `Approve`/`Reject`/`Open` inline-keyboard taps (a compact `apv|rej:<kind>:<id8>` callback codec; a reject reason or a task-approve note is collected via a force_reply prompt held in a TTL'd in-memory pending-action map) through the SAME CEO-gated service calls the HTTP routes make (task/release/xpost/video/roadmap), stamping a `via=telegram` audit row on each. Escalation DMs (not completion DMs) carry the actionable keyboard when V2 is armed. Closing the loop exposed a real hole: a stale Approve/Reject button targets its item by id regardless of current status, so `ReleaseProposalService.approve`/`.reject`, `XPostService.approve`, and `VideoPostService.approve` now all refuse an already-CANCELLED (rejected) or already-COMPLETED (published/posted) target instead of silently re-executing — a fix that also closes the identical hole via a replayed HTTP call, not just Telegram. V3 adds a Telegram **Mini App** sign-in: `POST /api/telegram/webapp-auth` (`roboco/api/routes/telegram.py`, mounted only when `telegram_miniapp_enabled` AND `cloud_auth_enabled` are both armed — `telegram_miniapp_enabled` is env-only like `cloud_auth_enabled`, deliberately off the panel feature-flags card, and fails loud at startup if armed without cloud auth on) validates Telegram's signed `initData` (`roboco/utils/telegram_initdata.py` — pure HMAC-SHA256 `WebAppData`-keyed validation, constant-time compare, a `telegram_initdata_max_age_seconds` freshness window with 60s clock-skew tolerance) against the stored bot token and the CEO's own `chat_id`, then mints the same cloud-auth session cookie `/api/auth/login` issues — turning the CEO's phone into a real panel client at the new `(tg)` route group (`/tg`: Approvals/Inbox/Board/Chat tabs, outside the normal dashboard shell; `proxy.ts`'s matcher excludes `tg(?:/|$)` so a phone session is never bounced to the password `/login` page it can't reach). Requires a public HTTPS origin (the cookie is secure-only) and BotFather's `/setmenubutton` pointed at `https://<host>/tg`.
+**Telegram notifications bridge V1+V2+V3 (default-off `ROBOCO_TELEGRAM_ENABLED`).** V1: best-effort, outbound-only Telegram DMs to the CEO on escalation and completion. Mirrors the `x_credentials` pattern: a singleton Fernet-encrypted `telegram_credentials` row (migration 074, bot token + chat id; the API returns `has_credentials` only) behind CEO-only `/telegram/credentials` routes and a panel credentials card. `_notify_telegram` (`roboco/services/notification_delivery.py`) fans out from `notify_ceo_of_escalation` / `notify_ceo_of_completion`, sending only the notification's subject plus an optional panel deep-link (`panel_base_url`) — never the body — via a deferred, best-effort send that never raises into the producer (`NullTelegramClient` when unconfigured or the flag is off, `LiveTelegramClient` posting to the Bot API otherwise). V2 (`ROBOCO_TELEGRAM_INBOUND_ENABLED`, sub-switch on top of V1's flag — both plus stored credentials are required, otherwise the bot only sends and never listens) makes the bridge two-way: `TelegramInboundEngine` (`roboco/services/telegram_inbound.py`) long-polls `getUpdates` from a dedicated orchestrator loop (`_telegram_poll_loop`), authorizing every update by BOTH chat id and sender id, and routes `/status` / `/queue` / `/task` commands plus `Approve`/`Reject`/`Open` inline-keyboard taps (a compact `apv|rej:<kind>:<id8>` callback codec; a reject reason or a task-approve note is collected via a force_reply prompt held in a TTL'd in-memory pending-action map) through the SAME CEO-gated service calls the HTTP routes make (task/release/xpost/video/roadmap), stamping a `via=telegram` audit row on each. Escalation DMs (not completion DMs) carry the actionable keyboard when V2 is armed. All bot/bridge messages are HTML-styled (`parse_mode=HTML` with mandatory `_esc`/`_esc_attr` escaping at every dynamic interpolation and balance-aware 4096 truncation — the injection posture moved from no-parse_mode to escaping discipline), and every held-draft origination (release proposal, X post, video post, roadmap item via `propose_roadmap`) pushes a styled DM with its Approve/Reject keyboard the moment it materializes (`notify_ceo_of_queue_item`, best-effort, sharing `/queue`'s renderer). Closing the loop exposed a real hole: a stale Approve/Reject button targets its item by id regardless of current status, so `ReleaseProposalService.approve`/`.reject`, `XPostService.approve`, and `VideoPostService.approve` now all refuse an already-CANCELLED (rejected) or already-COMPLETED (published/posted) target instead of silently re-executing — a fix that also closes the identical hole via a replayed HTTP call, not just Telegram. V3 adds a Telegram **Mini App** sign-in: `POST /api/telegram/webapp-auth` (`roboco/api/routes/telegram.py`, mounted only when `telegram_miniapp_enabled` AND `cloud_auth_enabled` are both armed — `telegram_miniapp_enabled` is env-only like `cloud_auth_enabled`, deliberately off the panel feature-flags card, and fails loud at startup if armed without cloud auth on) validates Telegram's signed `initData` (`roboco/utils/telegram_initdata.py` — pure HMAC-SHA256 `WebAppData`-keyed validation, constant-time compare, a `telegram_initdata_max_age_seconds` freshness window with 60s clock-skew tolerance) against the stored bot token and the CEO's own `chat_id`, then mints the same cloud-auth session cookie `/api/auth/login` issues — turning the CEO's phone into a real panel client at the new `(tg)` route group (`/tg`: Approvals/Inbox/Board/Chat tabs, outside the normal dashboard shell; `proxy.ts`'s matcher excludes `tg(?:/|$)` so a phone session is never bounced to the password `/login` page it can't reach). Requires a public HTTPS origin (the cookie is secure-only) and BotFather's `/setmenubutton` pointed at `https://<host>/tg`.
 
 **Possibilities matrix (default-off `ROBOCO_POSSIBILITIES_MATRIX_ENABLED`).** A work-already-done fast path on `i_am_done`: when a claimed/in_progress task already has commits, an open PR, every acceptance criterion addressed, and no open findings (`_work_appears_done`), the dev submits straight to QA in one call instead of the standard multi-turn plan/journal/local-gate derivation. `_i_am_done_fast_path` still runs the non-negotiable guards — ownership, branch-pushed, not-behind-base, conventions, `FINDINGS_ADDRESSED` — and trusts the PR's own CI-green signal as the quality-gate proxy (`_fast_path_quality_verdict`, the same signal `pr_pass` trusts); a repo with no CI signal falls back to the local `make quality` gate (plus the toolchain-match guard when `ROBOCO_TOOLCHAIN_MATCH_ENABLED` is armed), and a known-red CI refuses the fast path outright rather than shipping it to QA. The orchestrator's dev spawn prompt steers a matching task to a `WORK_ALREADY_DONE` state that tells the dev to call `i_am_done` directly instead of re-deriving what's already done.
 

docs/map/notification.md

@@ -153,6 +153,7 @@ telegram_inbound.py (TelegramInboundEngine, V2)
 | 3aff6e04 | Chore: Close gaps (#285) — follow-on gap closure touching notification.py / notification_dedup.py / notification_delivery.py | Refinement of the #283 changes (exact hunks not isolated per-file in this merge commit; consolidated the dedup/outbox behavior above) |
 
 > Post-snapshot updates (since 2026-06-29): 115061f3 fixed list_system_notifications pending_ack_only correctness: SQL limit is now dropped for that branch so newer fully-acked rows can't mask older unacked ones (see Gotcha update above). `61e00832` (PR #492) added `notify_auditor_of_rework()` and `_get_auditor_agent()` to power the reactive auditor dispatch path: HIGH-priority ALERT notifications addressed to the auditor agent are emitted when a task enters `needs_revision` via QA/PR/PM rework chokepoints. **Wave 3** (2026-07-17, PR #547): `CreateNotificationParams` gains `requires_ack: bool | None = None`, consulted in `_create_notification` ahead of the `ACK_REQUIRED_BY_TYPE` default; `send_a2a_notification` gains a `requires_ack: bool = False` kwarg (plus an `str | None` `task_id`, for a conversational DM with no task behind it) that threads through — the only caller passing True is `A2AService._maybe_wake_ceo_recipient` (docs/map/a2a-audit-journal-permissions.md), so its wake row is finally visible to the orchestrator's `_dispatch_a2a_work` `pending_ack_only` poll.
+> `cd978d11`+fixes (2026-07-18, wave-13): Telegram sends are HTML-styled — `_esc` (text nodes) / `_esc_attr` (href attributes) escaping discipline, balance-aware `_truncate`, `parse_mode`/`disable_link_preview` on the client; new `notify_ceo_of_queue_item` pushes a styled keyboard DM at each held-draft origination (release/x/video engines + `propose_roadmap`), sharing `telegram_inbound.render_queue_item_text`.
 > `3b9fd0e0`+`11915f36` (PR #551, Telegram V2): `3b9fd0e0` adds `telegram_inbound.py` (new file, `TelegramInboundEngine`), extends `telegram_client.py` with `get_updates`/`answer_callback_query`/`edit_message_reply_markup`/`edit_message_text`, adds `actionable=True` to `_notify_telegram` (escalation only) so the DM carries an Approve/Reject/Open keyboard, and wires the orchestrator's `_telegram_poll_loop`. `11915f36` closes a live-reproduced approve-after-reject hole reachable via a stale Telegram button (or the pre-existing HTTP routes for X/video): `ReleaseProposalService.approve()` now refuses CANCELLED (`already_rejected`) and COMPLETED (`already_published`) proposals via a new `_approve_precheck`, `.reject()` refuses COMPLETED by raising a new `TaskAlreadyCompletedError`, and `XPostService`/`VideoPostService.approve()` each add a CANCELLED pre-lock-and-under-lock guard returning `already_rejected`. Also adds `_authorized_sender` (chat-id auth is defense-in-depth'd with a sender-id check) and widens `_resolve_task`'s search limit 10→50 so a genuine id-prefix hit can't be pushed out by newer title/description matches.
 
 ## Regression Risks

roboco/services/gateway/content_actions.py

@@ -1344,6 +1344,7 @@ class ContentActions:
             task, {"goal": cycle_goal.strip(), "items": normalized}
         )
         await self.task.session.flush()
+        await self._notify_roadmap_items(task, normalized)
         return Envelope.ok(
             status="roadmap_proposed",
             task_id=str(task.id),
@@ -1354,6 +1355,30 @@ class ContentActions:
             },
         )
 
+    async def _notify_roadmap_items(
+        self, task: Any, items: list[dict[str, Any]]
+    ) -> None:
+        """Best-effort push DM per proposed item — this is the moment a
+        roadmap item first becomes CEO-actionable (the engine's own
+        exploration-task origination has nothing to review yet), so the DM
+        fires here rather than from ``RoadmapEngine``. A send failure never
+        blocks ``propose_roadmap`` itself."""
+        if self._deps.notification_delivery is None:
+            return
+        id8 = str(task.id)[:8]
+        for item in items:
+            try:
+                await self._deps.notification_delivery.notify_ceo_of_queue_item(
+                    kind="roadmap",
+                    id8=id8,
+                    extra=str(item.get("id") or ""),
+                    title=item.get("title") or "untitled",
+                )
+            except Exception as exc:
+                logger.warning(
+                    "roadmap telegram notify failed (best-effort)", error=str(exc)
+                )
+
     @classmethod
     def _reject_feature_spotlight_fields(
         cls, feature_slug: str, feature_title: str, body: str

roboco/services/notification_delivery.py

@@ -11,6 +11,7 @@ Also implements the ACK system for tracking acknowledgments.
 
 import asyncio
 import contextlib
+import html
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime
@@ -43,6 +44,22 @@ if TYPE_CHECKING:
 _log = structlog.get_logger(service="notification_delivery")
 
 
+def _esc(value: object) -> str:
+    """HTML-escape a dynamic value before it lands in a Telegram HTML
+    message — mirrors ``telegram_inbound._esc``; every DM this service
+    composes runs its dynamic parts (a notification subject, a panel link)
+    through this before interpolation."""
+    return html.escape(str(value), quote=False)
+
+
+def _esc_attr(value: object) -> str:
+    """Like ``_esc`` but also escapes quotes — mirrors
+    ``telegram_inbound._esc_attr``. The panel-link ``href`` is the one place
+    this service interpolates into an HTML attribute rather than a text
+    node; an unescaped ``"`` there would close the attribute early."""
+    return html.escape(str(value), quote=True)
+
+
 def _format_completion_body(task: TaskTable, metrics: "TaskMetrics | None") -> str:
     """Human-readable completion summary — real effort vs wall-clock, not a lone
     wall-clock figure. Degrades to wall-clock-only (turns 'n/a') when there are
@@ -891,26 +908,25 @@ class NotificationDeliveryService(BaseService):
             escalator_slug=escalator.slug,
         )
 
-    async def _notify_telegram(
+    async def _send_telegram_deferred(
-        self, *, task_id: UUID, subject: str, actionable: bool = False
+        self,
+        *,
+        text: str,
+        reply_markup: dict[str, Any] | None,
+        disable_link_preview: bool = False,
     ) -> None:
-        """Best-effort Telegram DM to the CEO alongside an in-app notification.
+        """Shared best-effort deferred-send plumbing behind every Telegram DM
+        this service issues (``_notify_telegram``, ``notify_ceo_of_queue_item``).
 
-        Degrades to a no-op unless ``telegram_enabled`` is armed and credentials
+        Degrades to a no-op unless ``telegram_enabled`` is armed and
-        are stored. Credentials are fetched now (a fast DB read on the open
+        credentials are stored. Credentials are fetched now (a fast DB read
-        session); the actual network send is deferred via
+        on the open session); the actual network send is deferred via
         ``defer_after_commit`` so a slow Telegram Bot API call can't hold the
         caller's open transaction for up to ``telegram_timeout_seconds``.
         Never raises into the caller — a credentials/network failure only
-        logs. The message carries a panel deep-link when ``panel_base_url``
+        logs. ``text`` is sent with HTML ``parse_mode``; callers are
-        is set.
+        responsible for escaping every dynamic value they interpolated into
-
+        it (``_esc``).
-        ``actionable=True`` (escalation only — V1's completion send never
-        expands beyond link-only, and no new call site is added here) also
-        attaches an Approve/Reject/Open inline keyboard (V2, gated separately
-        by ``telegram_inbound_enabled`` — with it off the buttons render but
-        the bot never polls for the tap, so they're harmlessly inert; the
-        plain-text link still works either way).
         """
         from roboco.config import settings
 
@@ -927,22 +943,18 @@ class NotificationDeliveryService(BaseService):
             _log.warning("telegram_notify_failed", error=str(exc))
             return
 
-        text = subject
-        if settings.panel_base_url:
-            link = f"{settings.panel_base_url.rstrip('/')}/tasks/{str(task_id)[:8]}"
-            text = f"{subject}\n{link}"
         timeout = settings.telegram_timeout_seconds
-        reply_markup = None
-        if actionable:
-            from roboco.services.telegram_inbound import build_action_keyboard
-
-            reply_markup = build_action_keyboard("task", str(task_id)[:8])
 
         async def _send() -> None:
             client = None
             try:
                 client = build_telegram_client(creds, timeout=timeout)
-                result = await client.send_message(text, reply_markup=reply_markup)
+                result = await client.send_message(
+                    text,
+                    reply_markup=reply_markup,
+                    parse_mode="HTML",
+                    disable_link_preview=disable_link_preview,
+                )
                 if not result.sent:
                     _log.warning("telegram_notify_skip", detail=result.detail)
             except Exception as exc:  # best-effort — never break the drain
@@ -954,6 +966,60 @@ class NotificationDeliveryService(BaseService):
 
         defer_after_commit(self.session, _send)
 
+    async def _notify_telegram(
+        self, *, task_id: UUID, subject: str, actionable: bool = False
+    ) -> None:
+        """Best-effort Telegram DM to the CEO alongside an in-app notification.
+
+        The message carries a panel deep-link (named "Open in panel", link
+        preview disabled so the card never swallows the chat) when
+        ``panel_base_url`` is set.
+
+        ``actionable=True`` (escalation only — V1's completion send never
+        expands beyond link-only, and no new call site is added here) also
+        attaches an Approve/Reject/Open inline keyboard (V2, gated separately
+        by ``telegram_inbound_enabled`` — with it off the buttons render but
+        the bot never polls for the tap, so they're harmlessly inert; the
+        plain-text link still works either way).
+        """
+        from roboco.config import settings
+
+        text = f"<b>{_esc(subject)}</b>"
+        if settings.panel_base_url:
+            link = f"{settings.panel_base_url.rstrip('/')}/tasks/{str(task_id)[:8]}"
+            text += f'\n<a href="{_esc_attr(link)}">Open in panel</a>'
+        reply_markup = None
+        if actionable:
+            from roboco.services.telegram_inbound import build_action_keyboard
+
+            reply_markup = build_action_keyboard("task", str(task_id)[:8])
+
+        await self._send_telegram_deferred(
+            text=text, reply_markup=reply_markup, disable_link_preview=True
+        )
+
+    async def notify_ceo_of_queue_item(
+        self, *, kind: str, id8: str, extra: str = "", title: str
+    ) -> None:
+        """Best-effort push DM at the moment a held draft becomes CEO-
+        actionable — release proposals, X drafts, video posts, and roadmap
+        items used to land in the approval queue silently, with no ping
+        until the CEO happened to run ``/queue``. Reuses the exact styled
+        item line and Approve/Reject/Open keyboard ``/queue`` itself renders
+        (``telegram_inbound.render_queue_item_text`` / ``build_action_keyboard``
+        — one renderer, two callers), and the same degrade-to-no-op contract
+        as ``_notify_telegram``: a credentials/network failure only logs,
+        never raises into the originating engine.
+        """
+        from roboco.services.telegram_inbound import (
+            build_action_keyboard,
+            render_queue_item_text,
+        )
+
+        text = render_queue_item_text(kind, id8, extra, title)
+        reply_markup = build_action_keyboard(kind, id8, extra)
+        await self._send_telegram_deferred(text=text, reply_markup=reply_markup)
+
     async def notify_ceo_of_escalation(
         self,
         *,

roboco/services/release_manager_engine.py

@@ -34,6 +34,7 @@ from roboco.foundation.policy.content import markers
 from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team
 from roboco.services.base import BaseService
 from roboco.services.notification import NotificationService
+from roboco.services.notification_delivery import get_notification_delivery_service
 from roboco.services.project import get_project_service
 from roboco.services.release_readiness import (
     ReleaseReadinessReport,
@@ -204,6 +205,18 @@ class ReleaseManagerEngine(BaseService):
             )
         except Exception as exc:
             self.log.warning("release CEO notify failed (best-effort)", error=str(exc))
+        try:
+            await get_notification_delivery_service(
+                self.session
+            ).notify_ceo_of_queue_item(
+                kind="release",
+                id8=str(task.id)[:8],
+                title=f"v{report.proposed_version} ready",
+            )
+        except Exception as exc:
+            self.log.warning(
+                "release telegram notify failed (best-effort)", error=str(exc)
+            )
 
     async def _production_assess(self) -> ReleaseReadinessReport | None:
         """Real path: read-clone RoboCo, fetch CI, gather the snapshot, assess.

roboco/services/telegram_client.py

@@ -18,10 +18,13 @@ from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
 import httpx
+import structlog
 
 if TYPE_CHECKING:
     from roboco.services.telegram_credentials import TelegramCredentialsData
 
+logger = structlog.get_logger()
+
 _API_BASE = "https://api.telegram.org"
 
 
@@ -51,6 +54,8 @@ class TelegramClient(ABC):
         *,
         reply_markup: dict[str, Any] | None = None,
         reply_to_message_id: int | None = None,
+        parse_mode: str | None = None,
+        disable_link_preview: bool = False,
     ) -> TelegramSendResult: ...
 
     @abstractmethod
@@ -75,7 +80,14 @@ class TelegramClient(ABC):
     ) -> None: ...
 
     @abstractmethod
-    async def edit_message_text(self, message_id: int, text: str) -> None: ...
+    async def edit_message_text(
+        self,
+        message_id: int,
+        text: str,
+        *,
+        parse_mode: str | None = None,
+        disable_link_preview: bool = False,
+    ) -> None: ...
 
     @abstractmethod
     async def close(self) -> None:
@@ -98,8 +110,10 @@ class NullTelegramClient(TelegramClient):
         *,
         reply_markup: dict[str, Any] | None = None,
         reply_to_message_id: int | None = None,
+        parse_mode: str | None = None,
+        disable_link_preview: bool = False,
     ) -> TelegramSendResult:
-        _ = (text, reply_markup, reply_to_message_id)
+        _ = (text, reply_markup, reply_to_message_id, parse_mode, disable_link_preview)
         return TelegramSendResult(sent=False, detail="no credentials configured")
 
     async def get_updates(
@@ -118,8 +132,15 @@ class NullTelegramClient(TelegramClient):
     ) -> None:
         _ = (message_id, reply_markup)
 
-    async def edit_message_text(self, message_id: int, text: str) -> None:
+    async def edit_message_text(
-        _ = (message_id, text)
+        self,
+        message_id: int,
+        text: str,
+        *,
+        parse_mode: str | None = None,
+        disable_link_preview: bool = False,
+    ) -> None:
+        _ = (message_id, text, parse_mode, disable_link_preview)
 
 
 class LiveTelegramClient(TelegramClient):
@@ -151,12 +172,31 @@ class LiveTelegramClient(TelegramClient):
             await self._client.aclose()
             self._client = None
 
+    @staticmethod
+    def _format_payload(
+        payload: dict[str, Any],
+        *,
+        parse_mode: str | None,
+        disable_link_preview: bool,
+    ) -> None:
+        """Mutates ``payload`` in place with the formatting fields shared by
+        ``sendMessage``/``editMessageText`` — only when actually set, so a
+        plain-text caller's payload is byte-for-byte unchanged. Bot API 7.0+
+        field name (``disable_web_page_preview`` is the legacy alias, still
+        accepted, but ``link_preview_options`` is current)."""
+        if parse_mode:
+            payload["parse_mode"] = parse_mode
+        if disable_link_preview:
+            payload["link_preview_options"] = {"is_disabled": True}
+
     async def send_message(
         self,
         text: str,
         *,
         reply_markup: dict[str, Any] | None = None,
         reply_to_message_id: int | None = None,
+        parse_mode: str | None = None,
+        disable_link_preview: bool = False,
     ) -> TelegramSendResult:
         url = f"{_API_BASE}/bot{self._creds.bot_token}/sendMessage"
         payload: dict[str, Any] = {"chat_id": self._creds.chat_id, "text": text}
@@ -164,6 +204,9 @@ class LiveTelegramClient(TelegramClient):
             payload["reply_markup"] = reply_markup
         if reply_to_message_id is not None:
             payload["reply_to_message_id"] = reply_to_message_id
+        self._format_payload(
+            payload, parse_mode=parse_mode, disable_link_preview=disable_link_preview
+        )
         client = await self._http()
         try:
             resp = await client.post(url, json=payload, timeout=self._timeout)
@@ -223,16 +266,32 @@ class LiveTelegramClient(TelegramClient):
         with contextlib.suppress(httpx.HTTPError):
             await client.post(url, json=payload, timeout=self._timeout)
 
-    async def edit_message_text(self, message_id: int, text: str) -> None:
+    async def edit_message_text(
+        self,
+        message_id: int,
+        text: str,
+        *,
+        parse_mode: str | None = None,
+        disable_link_preview: bool = False,
+    ) -> None:
         url = f"{_API_BASE}/bot{self._creds.bot_token}/editMessageText"
         payload: dict[str, Any] = {
             "chat_id": self._creds.chat_id,
             "message_id": message_id,
             "text": text,
         }
+        self._format_payload(
+            payload, parse_mode=parse_mode, disable_link_preview=disable_link_preview
+        )
         client = await self._http()
         with contextlib.suppress(httpx.HTTPError):
-            await client.post(url, json=payload, timeout=self._timeout)
+            resp = await client.post(url, json=payload, timeout=self._timeout)
+            if not resp.is_success:
+                logger.warning(
+                    "telegram edit_message_text failed",
+                    status_code=resp.status_code,
+                    detail=resp.text[:200],
+                )
 
 
 def build_telegram_client(

roboco/services/telegram_inbound.py

@@ -22,15 +22,27 @@ again).
 The getUpdates offset cursor persists in the existing ``system_settings`` KV
 store (``telegram_last_update_id``) rather than a new table, so a restart
 doesn't replay already-processed updates.
+
+Every outbound send uses Telegram's HTML ``parse_mode`` for real hierarchy
+(bold headers, ``<code>`` ids, named links) instead of the original flat
+plain-text posture. That only stays injection-safe because EVERY dynamic
+value — task titles, reasons, subjects, urls, team/kind names — is run
+through ``_esc`` (``html.escape``) before it is interpolated; the only
+unescaped HTML in any composed message is the static markup this module
+writes itself. Telegram-HTML supports only a small tag subset (b/i/u/s/code/
+pre/a/blockquote) — nothing else is ever emitted.
 """
 
 from __future__ import annotations
 
+import html
+import re
 import time
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, cast
 from uuid import UUID
 
+import structlog
 from sqlalchemy import func, select
 
 from roboco.config import settings
@@ -70,6 +82,8 @@ if TYPE_CHECKING:
     from roboco.db.tables import TaskTable
     from roboco.services.telegram_credentials import TelegramCredentialsData
 
+logger = structlog.get_logger()
+
 # Reuses the system_settings KV store instead of a dedicated table (see
 # `roboco.services.settings._VALIDATORS` for the write-side int validator).
 _OFFSET_KEY = "telegram_last_update_id"
@@ -91,19 +105,118 @@ _DEFAULT_REJECT_MIN_CHARS = (
 # trail requirement for a CEO approval note.
 _TASK_APPROVE_MIN_CHARS = 20
 
+# /status render order — the lifecycle's actual flow (pending -> claimed ->
+# in_progress -> paused|blocked -> the review-gate chain -> completed /
+# cancelled), not TaskStatus's declaration order or an alphabetical dump.
+_STATUS_ORDER: tuple[TaskStatus, ...] = (
+    TaskStatus.BACKLOG,
+    TaskStatus.PENDING,
+    TaskStatus.CLAIMED,
+    TaskStatus.IN_PROGRESS,
+    TaskStatus.PAUSED,
+    TaskStatus.BLOCKED,
+    TaskStatus.VERIFYING,
+    TaskStatus.NEEDS_REVISION,
+    TaskStatus.AWAITING_QA,
+    TaskStatus.AWAITING_DOCUMENTATION,
+    TaskStatus.AWAITING_PR_REVIEW,
+    TaskStatus.AWAITING_PM_REVIEW,
+    TaskStatus.AWAITING_CEO_APPROVAL,
+    TaskStatus.COMPLETED,
+    TaskStatus.CANCELLED,
+)
+
+# /queue + push-DM item rendering — (emoji, label) per kind.
+_KIND_DISPLAY: dict[str, tuple[str, str]] = {
+    "release": ("🚀", "Release"),
+    "video": ("🎬", "Video"),
+    "xpost": ("✕", "Post"),
+    "roadmap": ("🗺️", "Roadmap"),
+    "task": ("📋", "Task"),
+}
+
 _HELP_TEXT = (
-    "Commands:\n"
+    "<b>RoboCo commands</b>\n"
-    "/status - fleet snapshot\n"
+    "/status — fleet snapshot\n"
-    "/queue - everything awaiting your approval, with buttons\n"
+    "/queue — approvals with one-tap buttons\n"
-    "/task <id8-or-title> - task detail\n"
+    "/task — task detail (id prefix or title)\n"
-    "/help - this message"
+    "/help — this list"
 )
 
 
+def _esc(value: object) -> str:
+    """HTML-escape any dynamic value before it lands in a Telegram HTML
+    message. Telegram-HTML has no safe-interpolation primitive of its own —
+    every dynamic string (task titles, reasons, URLs, ...) must be escaped by
+    hand before assembly. Static markup this module writes is the only
+    unescaped HTML."""
+    return html.escape(str(value), quote=False)
+
+
+def _esc_attr(value: object) -> str:
+    """Like ``_esc`` but also escapes quotes (``quote=True``). Every
+    ``href="..."`` value sits inside an HTML attribute, not a text node — an
+    unescaped ``"`` in the value closes the attribute early and lets the rest
+    of the string inject arbitrary markup/attributes into the ``<a>`` tag."""
+    return html.escape(str(value), quote=True)
+
+
+# Telegram-HTML's own supported tag subset (see module docstring) — the only
+# tags ``_truncate`` ever needs to balance, since every dynamic value is
+# escaped before assembly.
+_TELEGRAM_TAGS = frozenset({"b", "i", "u", "s", "code", "pre", "a", "blockquote"})
+_TAG_RE = re.compile(r"</?([a-z]+)(?:\s[^>]*)?>")
+
+
+def _safe_boundary(cut: str) -> str:
+    """Back a truncated slice off the last mid-tag (``<b>...<``) or
+    mid-entity (``&am``) boundary — Telegram's HTML parser rejects either
+    outright."""
+    last_lt, last_gt = cut.rfind("<"), cut.rfind(">")
+    if last_lt > last_gt:  # an unclosed '<...' — back off before it
+        cut = cut[:last_lt]
+    last_amp, last_semi = cut.rfind("&"), cut.rfind(";")
+    if last_amp > last_semi:  # an unclosed '&...' entity — back off before it
+        cut = cut[:last_amp]
+    return cut
+
+
+def _closing_tags(text: str) -> str:
+    """The Telegram-HTML tags still open at the end of ``text``, rendered as
+    closing tags in reverse (innermost-first) order."""
+    stack: list[str] = []
+    for m in _TAG_RE.finditer(text):
+        name = m.group(1)
+        if name not in _TELEGRAM_TAGS:
+            continue
+        if m.group(0).startswith("</"):
+            if stack and stack[-1] == name:
+                stack.pop()
+        else:
+            stack.append(name)
+    return "".join(f"</{name}>" for name in reversed(stack))
+
+
 def _truncate(text: str, limit: int = _MESSAGE_CHAR_LIMIT) -> str:
+    """Telegram's own sendMessage cap, applied to the FINAL assembled HTML
+    string. Every dynamic value is escaped before assembly (``_esc``), so the
+    only tags/entities present are the small fixed set this module emits —
+    but a naive slice can still land mid-tag, mid-entity, or (since the slice
+    point falls wherever the char count happens to land) inside an as-yet-
+    unclosed tag like ``<code>``, and Telegram's HTML parser rejects an
+    unbalanced message outright. Back off to the last safe tag/entity
+    boundary (``_safe_boundary``), then append whatever closing tags are
+    still owed (``_closing_tags``) — trimming further first if the closing
+    tags themselves would push past ``limit``.
+    """
     if len(text) <= limit:
         return text
-    return text[: limit - 1].rstrip() + "…"
+    cut = _safe_boundary(text[: limit - 1])
+    closing = _closing_tags(cut)
+    while len(cut.rstrip()) + 1 + len(closing) > limit:
+        cut = _safe_boundary(cut[:-1])
+        closing = _closing_tags(cut)
+    return cut.rstrip() + "…" + closing
 
 
 def parse_command(text: str) -> tuple[str, str]:
@@ -159,6 +272,20 @@ def parse_callback(data: str) -> ParsedCallback | None:
     return ParsedCallback(action=action, kind=kind, id8=id8, extra=extra)
 
 
+def render_queue_item_text(kind: str, id8: str, extra: str, title: str) -> str:
+    """One styled ``<emoji> <b>Kind</b> — <escaped title> <code>id8[:extra]</code>``
+    line. Shared by ``/queue``'s listing (``_send_queue``) and the
+    origination-time push DM (``NotificationDeliveryService.
+    notify_ceo_of_queue_item``) so a freshly-drafted item and the same item
+    later listed by ``/queue`` render identically — one renderer, two
+    callers."""
+    emoji, label = _KIND_DISPLAY.get(kind, ("📋", kind.title()))
+    suffix = f":{extra}" if extra else ""
+    return _truncate(
+        f"{emoji} <b>{label}</b> — {_esc(title)} <code>{_esc(id8 + suffix)}</code>"
+    )
+
+
 # Deep-link target per kind — mirrors where each queue actually lives in the
 # panel (release proposal + roadmap both surface on the Overview command
 # center; X/video drafts on the Social queue).
@@ -330,7 +457,9 @@ class TelegramInboundEngine(BaseService):
                 if pending.expires_at > time.monotonic():
                     await self._consume_reply(pending, text, client)
                     return
-                await client.send_message("That prompt expired — tap the button again.")
+                await client.send_message(
+                    "That prompt expired — tap the button again.", parse_mode="HTML"
+                )
                 return
         cmd, args = parse_command(text)
         if not cmd:
@@ -341,19 +470,28 @@ class TelegramInboundEngine(BaseService):
         self, cmd: str, args: str, client: TelegramClient
     ) -> None:
         if cmd in ("start", "help"):
-            await client.send_message(_HELP_TEXT)
+            await client.send_message(_HELP_TEXT, parse_mode="HTML")
         elif cmd == "status":
-            await client.send_message(await self._render_status())
+            await client.send_message(await self._render_status(), parse_mode="HTML")
         elif cmd == "queue":
             await self._send_queue(client)
         elif cmd == "task":
-            await client.send_message(await self._render_task(args))
+            await client.send_message(
+                await self._render_task(args),
+                parse_mode="HTML",
+                disable_link_preview=True,
+            )
         else:
-            await client.send_message(f"Unknown command /{cmd}.\n\n{_HELP_TEXT}")
+            await client.send_message(
+                f"Unknown command /{_esc(cmd)}.\n\n{_HELP_TEXT}", parse_mode="HTML"
+            )
 
     async def _render_status(self) -> str:
         """Cheap snapshot: active-agent count + task counts by status — no
-        spend/strategy/pitch queries (that's the heavier cockpit summary)."""
+        spend/strategy/pitch queries (that's the heavier cockpit summary).
+        Statuses render in lifecycle order (only the nonzero ones), not
+        alphabetically — a CEO scanning on a phone reads top-to-bottom as the
+        pipeline, not as an a-z dump."""
         counts = await get_task_service(self.session).count_by_status()
         active_result = await self.session.execute(
             select(func.count(AgentTable.id)).where(
@@ -361,8 +499,17 @@ class TelegramInboundEngine(BaseService):
             )
         )
         active = active_result.scalar_one()
-        lines = [f"Active agents: {active}", "", "Tasks by status:"]
+        lines = [
-        lines += [f"  {k}: {v}" for k, v in sorted(counts.items())]
+            "<b>🤖 Fleet</b>",
+            f"Active agents: <b>{active}</b>",
+            "",
+            "<b>📋 Tasks</b>",
+        ]
+        lines += [
+            f"• {status.value} — <b>{counts[status.value]}</b>"
+            for status in _STATUS_ORDER
+            if counts.get(status.value)
+        ]
         return _truncate("\n".join(lines))
 
     async def _collect_queue_items(self) -> list[tuple[str, str, str, str]]:
@@ -379,10 +526,7 @@ class TelegramInboundEngine(BaseService):
 
     async def _queue_items_for_tasks(self) -> list[tuple[str, str, str, str]]:
         tasks = await get_task_service(self.session).list_awaiting_ceo_approval()
-        return [
+        return [("task", str(t.id)[:8], "", t.title or "Untitled") for t in tasks]
-            ("task", str(t.id)[:8], "", f"[Task] {t.title or 'Untitled'}")
-            for t in tasks
-        ]
 
     async def _queue_items_for_release(self) -> list[tuple[str, str, str, str]]:
         proposal = await get_release_proposal_service(self.session).open_proposal()
@@ -391,7 +535,7 @@ class TelegramInboundEngine(BaseService):
         id8 = str(proposal.id)[:8]
         report = markers.get_release_report(proposal) or {}
         version = report.get("proposed_version") or "?"
-        return [("release", id8, "", f"[Release] v{version} ready")]
+        return [("release", id8, "", f"v{version} ready")]
 
     async def _queue_items_for_xposts(self) -> list[tuple[str, str, str, str]]:
         posts = await get_x_post_service(self.session).list_open_posts()
@@ -399,7 +543,7 @@ class TelegramInboundEngine(BaseService):
         for post in posts:
             id8 = str(post.id)[:8]
             body = markers.get_x_draft_body(post) or post.description or ""
-            result.append(("xpost", id8, "", f"[X post] {body[:100]}"))
+            result.append(("xpost", id8, "", body[:100]))
         return result
 
     async def _queue_items_for_videos(self) -> list[tuple[str, str, str, str]]:
@@ -409,7 +553,7 @@ class TelegramInboundEngine(BaseService):
             id8 = str(post.id)[:8]
             draft = markers.get_video_draft(post) or {}
             occasion = draft.get("occasion") or "untitled"
-            result.append(("video", id8, "", f"[Video] {occasion}"))
+            result.append(("video", id8, "", occasion))
         return result
 
     async def _queue_items_for_roadmap(self) -> list[tuple[str, str, str, str]]:
@@ -430,21 +574,34 @@ class TelegramInboundEngine(BaseService):
                 continue
             item_id = str(item.get("id") or "")
             title = item.get("title") or "untitled"
-            result.append(("roadmap", id8, item_id, f"[Roadmap] {title}"))
+            result.append(("roadmap", id8, item_id, title))
         return result
 
     async def _send_queue(self, client: TelegramClient) -> None:
         items = await self._collect_queue_items()
         if not items:
-            await client.send_message("Nothing awaiting your approval.")
-            return
-        await client.send_message(f"{len(items)} item(s) awaiting your approval:")
-        for kind, id8, extra, title in items[:_QUEUE_ITEM_CAP]:
-            suffix = f":{extra}" if extra else ""
-            text = _truncate(f"{title} ({id8}{suffix})")
             await client.send_message(
-                text, reply_markup=build_action_keyboard(kind, id8, extra)
+                "✅ Nothing awaiting your approval.", parse_mode="HTML"
             )
+            return
+        noun = "item" if len(items) == 1 else "items"
+        await client.send_message(
+            f"<b>🔔 Awaiting your approval</b> — {len(items)} {noun}",
+            parse_mode="HTML",
+        )
+        for kind, id8, extra, title in items[:_QUEUE_ITEM_CAP]:
+            result = await client.send_message(
+                render_queue_item_text(kind, id8, extra, title),
+                reply_markup=build_action_keyboard(kind, id8, extra),
+                parse_mode="HTML",
+            )
+            if not result.sent:
+                logger.warning(
+                    "telegram queue item send failed",
+                    kind=kind,
+                    id8=id8,
+                    detail=result.detail,
+                )
 
     async def _resolve_task(self, id8: str) -> TaskTable | None:
         """Exact id-prefix resolution — ``search_tasks`` also OR-matches
@@ -464,7 +621,7 @@ class TelegramInboundEngine(BaseService):
     async def _render_task(self, args: str) -> str:
         q = args.strip()
         if not q:
-            return "Usage: /task <id8-or-title-fragment>"
+            return "Usage: /task id8-or-title-fragment"
         task = await self._resolve_task(q)
         if task is None:
             resolved_or_error = await self._search_task_by_fragment(q)
@@ -479,11 +636,14 @@ class TelegramInboundEngine(BaseService):
         rendered error/disambiguation message for the caller to return as-is."""
         candidates = await get_task_service(self.session).search_tasks(q, limit=5)
         if not candidates:
-            return f"No task matches {q!r}."
+            return f"No task matches {_esc(q)!r}."
         if len(candidates) > 1:
-            listing = "\n".join(f"  {str(t.id)[:8]} - {t.title}" for t in candidates)
+            listing = "\n".join(
+                f"• <code>{_esc(str(t.id)[:8])}</code> — {_esc(t.title)}"
+                for t in candidates
+            )
             return (
-                f"Multiple matches for {q!r}:\n{listing}\n\n"
+                f"Multiple matches for {_esc(q)!r}:\n{listing}\n\n"
                 "Retry with a more specific id or title."
             )
         return candidates[0]
@@ -497,15 +657,15 @@ class TelegramInboundEngine(BaseService):
             task.team.value if task.team and hasattr(task.team, "value") else "n/a"
         )
         lines = [
-            f"[{id8}] {task.title or 'Untitled'}",
+            f"<b><code>{_esc(id8)}</code> {_esc(task.title or 'Untitled')}</b>",
-            f"Status: {status_val}",
+            f"Status: <b>{_esc(status_val)}</b>",
-            f"Team: {team_val}",
+            f"Team: {_esc(team_val)}",
         ]
         if task.pr_url:
-            lines.append(f"PR: {task.pr_url}")
+            lines.append(f'PR: <a href="{_esc_attr(task.pr_url)}">View PR</a>')
         link = _deep_link("task", id8)
         if link:
-            lines.append(link)
+            lines.append(f'<a href="{_esc_attr(link)}">Open in panel</a>')
         return "\n".join(lines)
 
     # ---- callback (button) handling ---------------------------------------
@@ -553,7 +713,7 @@ class TelegramInboundEngine(BaseService):
         client: TelegramClient,
     ) -> None:
         field = (
-            f"approval notes (>= {_TASK_APPROVE_MIN_CHARS} chars)"
+            f"approval notes (at least {_TASK_APPROVE_MIN_CHARS} chars)"
             if parsed.action == "apv"
             else "rejection reason"
         )
@@ -561,8 +721,9 @@ class TelegramInboundEngine(BaseService):
             f":{parsed.extra}" if parsed.extra else ""
         )
         prompt = await client.send_message(
-            f"Reply to THIS message with your {field} for {target}.",
+            f"Reply to THIS message with your {field} for <code>{_esc(target)}</code>.",
             reply_markup={"force_reply": True},
+            parse_mode="HTML",
         )
         if prompt.message_id is None:
             return
@@ -596,14 +757,20 @@ class TelegramInboundEngine(BaseService):
     ) -> None:
         """Clears the original message's button row and stamps the outcome —
         the chat stays honest instead of leaving a stale, still-clickable
-        Approve/Reject row after the action already happened."""
+        Approve/Reject row after the action already happened. ``text`` is the
+        single funnel every approve/reject outcome (across all five kinds)
+        flows through, so it's escaped once here rather than at each of the
+        dozen call sites that compose it — it may embed a task title, a
+        rejection reason, or a service-raised error message, all dynamic."""
         icon = "✅" if ok else "❌"
-        rendered = _truncate(f"{icon} {text}")
+        rendered = _truncate(f"{icon} {_esc(text)}")
         if origin_message_id is not None:
             await client.edit_message_reply_markup(int(origin_message_id), None)
-            await client.edit_message_text(int(origin_message_id), rendered)
+            await client.edit_message_text(
+                int(origin_message_id), rendered, parse_mode="HTML"
+            )
         else:
-            await client.send_message(rendered)
+            await client.send_message(rendered, parse_mode="HTML")
 
     def _mark_audit(
         self, kind: str, task_id: UUID, action: str, *, item_id: str = ""

roboco/services/video_engine.py

@@ -29,6 +29,7 @@ from roboco.foundation.policy.content import markers
 from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team
 from roboco.services.base import BaseService
 from roboco.services.company_goals import get_company_goals_service
+from roboco.services.notification_delivery import get_notification_delivery_service
 from roboco.services.project import get_project_service
 from roboco.services.task import (
     VIDEO_POST_SOURCE,
@@ -534,6 +535,16 @@ class VideoEngine(BaseService):
             "video-engine: video post drafted (held for CEO)",
             source_task_id=str(source_task.id),
         )
+        try:
+            await get_notification_delivery_service(
+                self.session
+            ).notify_ceo_of_queue_item(
+                kind="video", id8=str(task.id)[:8], title=occasion
+            )
+        except Exception as exc:
+            self.log.warning(
+                "video-engine: telegram notify failed (best-effort)", error=str(exc)
+            )
         return task
 
     # ---- reject -> re-author (CEO feedback loop) ---------------------------

roboco/services/x_engine.py

@@ -49,6 +49,7 @@ from roboco.foundation.policy.injection_guard import screen_external_text
 from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team
 from roboco.services.base import BaseService
 from roboco.services.company_goals import get_company_goals_service
+from roboco.services.notification_delivery import get_notification_delivery_service
 from roboco.services.project import get_project_service
 from roboco.services.task import (
     X_FEATURE_EXPLORATION_SOURCE,
@@ -771,7 +772,12 @@ class XEngine(BaseService):
     async def _originate_post(
         self, *, title: str, body: str, source: str, project_id: UUID
     ) -> TaskTable:
-        """Open ONE PENDING, HELD X draft owned by the Secretary."""
+        """Open ONE PENDING, HELD X draft owned by the Secretary.
+
+        The single chokepoint for all three X sources (release post, mention
+        reply, feature spotlight — the last via ``materialize_feature_spotlight``
+        below) — one push-DM call here covers all three.
+        """
         task_svc = get_task_service(self.session)
         task = await task_svc.create(
             TaskCreateRequest(
@@ -792,6 +798,16 @@ class XEngine(BaseService):
         )
         markers.set_x_draft_body(task, body)
         await self.session.flush()
+        try:
+            await get_notification_delivery_service(
+                self.session
+            ).notify_ceo_of_queue_item(
+                kind="xpost", id8=str(task.id)[:8], title=body[:100]
+            )
+        except Exception as exc:
+            self.log.warning(
+                "x-engine: telegram notify failed (best-effort)", error=str(exc)
+            )
         return task
 
     async def materialize_feature_spotlight(

tests/integration/test_notification_delivery_phantom.py

@@ -290,8 +290,10 @@ async def test_notify_telegram_send_deferred_to_after_commit(
             *,
             reply_markup: dict | None = None,
             reply_to_message_id: int | None = None,
+            parse_mode: str | None = None,
+            disable_link_preview: bool = False,
         ) -> TelegramSendResult:
-            _ = (reply_markup, reply_to_message_id)
+            _ = (reply_markup, reply_to_message_id, parse_mode, disable_link_preview)
             sent.append(text)
             return TelegramSendResult(sent=True)
 
@@ -312,7 +314,73 @@ async def test_notify_telegram_send_deferred_to_after_commit(
     await db_session.commit()
     await _await_drain(db_session)
 
-    assert sent == ["Hello CEO"]
+    assert sent == ["<b>Hello CEO</b>"]
+
+
+@pytest.mark.asyncio
+async def test_notify_ceo_of_queue_item_deferred_escaped_and_keyboarded(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The origination-time push DM (release/xpost/video/roadmap drafts)
+    reuses the exact ``/queue`` item renderer + keyboard, rides the same
+    after-commit outbox as ``_notify_telegram``, and escapes a malicious
+    title before it ever reaches the Bot API payload."""
+    monkeypatch.setattr(settings, "telegram_enabled", True)
+
+    creds = TelegramCredentialsData(bot_token="t", chat_id="1")
+
+    class _FakeCredsService:
+        async def get_decrypted(self) -> TelegramCredentialsData:
+            return creds
+
+    monkeypatch.setattr(
+        "roboco.services.telegram_credentials.get_telegram_credentials_service",
+        lambda _session: _FakeCredsService(),
+    )
+
+    sent: list[tuple[str, dict | None, str | None]] = []
+
+    class _FakeTelegramClient:
+        async def send_message(
+            self,
+            text: str,
+            *,
+            reply_markup: dict | None = None,
+            reply_to_message_id: int | None = None,
+            parse_mode: str | None = None,
+            disable_link_preview: bool = False,
+        ) -> TelegramSendResult:
+            _ = (reply_to_message_id, disable_link_preview)
+            sent.append((text, reply_markup, parse_mode))
+            return TelegramSendResult(sent=True)
+
+        async def close(self) -> None:
+            pass
+
+    monkeypatch.setattr(
+        "roboco.services.telegram_client.build_telegram_client",
+        lambda _creds, **_kwargs: _FakeTelegramClient(),
+    )
+
+    service = get_notification_delivery_service(db_session)
+    await service.notify_ceo_of_queue_item(
+        kind="release", id8="a1b2c3d4", title="<b>v1.0.0</b> ready"
+    )
+
+    assert sent == []  # deferred — nothing before commit
+
+    await db_session.commit()
+    await _await_drain(db_session)
+
+    assert len(sent) == 1
+    text, reply_markup, parse_mode = sent[0]
+    assert "&lt;b&gt;v1.0.0&lt;/b&gt; ready" in text
+    assert "<b>v1.0.0</b> ready" not in text  # never unescaped
+    assert text.startswith("🚀 <b>Release</b>")
+    assert parse_mode == "HTML"
+    assert reply_markup is not None
+    row = reply_markup["inline_keyboard"][0]
+    assert row[0]["callback_data"] == "apv:release:a1b2c3d4"
 
 
 @pytest.mark.asyncio
@@ -344,8 +412,10 @@ async def test_notify_telegram_rollback_drops_send(
             *,
             reply_markup: dict | None = None,
             reply_to_message_id: int | None = None,
+            parse_mode: str | None = None,
+            disable_link_preview: bool = False,
         ) -> TelegramSendResult:
-            _ = (reply_markup, reply_to_message_id)
+            _ = (reply_markup, reply_to_message_id, parse_mode, disable_link_preview)
             sent.append(text)
             return TelegramSendResult(sent=True)
 

tests/unit/gateway/test_content_actions_roadmap.py

@@ -28,7 +28,7 @@ class _FakeTask:
         self.orchestration_markers = orchestration_markers
 
 
-def _actions(role: str) -> ContentActions:
+def _actions(role: str, *, notification_delivery: Any = None) -> ContentActions:
     task = MagicMock()
     agent = MagicMock()
     agent.role = role
@@ -41,6 +41,7 @@ def _actions(role: str) -> ContentActions:
         journal=MagicMock(),
         workspace=MagicMock(),
         notifications=MagicMock(),
+        notification_delivery=notification_delivery,
     )
     return ContentActions(deps)
 
@@ -175,6 +176,64 @@ async def test_propose_roadmap_persists_cycle_onto_open_task(
     actions.task.session.flush.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_propose_roadmap_sends_telegram_push_per_item(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Roadmap items only become CEO-actionable once propose_roadmap lands
+    (the engine's exploration-task origination has nothing to review yet),
+    so the push DM fires once per item here, not from RoadmapEngine."""
+    monkeypatch.setattr(cfg, "roadmap_min_items_per_cycle", 1)
+    monkeypatch.setattr(cfg, "roadmap_max_items_per_cycle", 7)
+    agent_id = uuid4()
+    cycle_task = _FakeTask(assigned_to=agent_id)
+    task_svc = MagicMock()
+    task_svc.list_open_roadmap_cycles = AsyncMock(return_value=[cycle_task])
+    monkeypatch.setattr("roboco.services.task.get_task_service", lambda _s: task_svc)
+    notify = AsyncMock()
+    actions = _actions("product_owner", notification_delivery=notify)
+    actions.task.session.flush = AsyncMock()
+
+    items = _valid_items(3)
+    env = await actions.propose_roadmap(
+        agent_id=agent_id, cycle_goal="Close onboarding friction", items=items
+    )
+    assert env.error is None
+
+    assert notify.notify_ceo_of_queue_item.await_count == len(items)
+    id8 = str(cycle_task.id)[:8]
+    for i, call in enumerate(notify.notify_ceo_of_queue_item.await_args_list):
+        assert call.kwargs["kind"] == "roadmap"
+        assert call.kwargs["id8"] == id8
+        assert call.kwargs["extra"] == f"item-{i}"
+        assert call.kwargs["title"] == f"Item {i}"
+
+
+@pytest.mark.asyncio
+async def test_propose_roadmap_survives_telegram_push_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A Telegram send failure must never block propose_roadmap itself."""
+    monkeypatch.setattr(cfg, "roadmap_min_items_per_cycle", 1)
+    monkeypatch.setattr(cfg, "roadmap_max_items_per_cycle", 7)
+    agent_id = uuid4()
+    cycle_task = _FakeTask(assigned_to=agent_id)
+    task_svc = MagicMock()
+    task_svc.list_open_roadmap_cycles = AsyncMock(return_value=[cycle_task])
+    monkeypatch.setattr("roboco.services.task.get_task_service", lambda _s: task_svc)
+    notify = MagicMock()
+    notify.notify_ceo_of_queue_item = AsyncMock(side_effect=RuntimeError("boom"))
+    actions = _actions("product_owner", notification_delivery=notify)
+    actions.task.session.flush = AsyncMock()
+
+    env = await actions.propose_roadmap(
+        agent_id=agent_id, cycle_goal="Close onboarding friction", items=_valid_items(2)
+    )
+
+    assert env.error is None
+    assert env.status == "roadmap_proposed"
+
+
 @pytest.mark.asyncio
 async def test_propose_roadmap_ignores_cycle_assigned_to_another_agent(
     monkeypatch: pytest.MonkeyPatch,

tests/unit/services/test_release_manager_engine.py

@@ -219,6 +219,49 @@ async def test_loop_never_publishes_or_approves(
     assert proposals[0].status == TS.PENDING  # never advanced by the loop
 
 
+@pytest.mark.asyncio
+async def test_proposes_sends_telegram_push(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A freshly-originated release proposal fires the styled push DM
+    (release kind, the proposal's id8, its version) alongside the existing
+    in-app notification."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    notify = AsyncMock()
+    monkeypatch.setattr(
+        "roboco.services.notification_delivery.NotificationDeliveryService."
+        "notify_ceo_of_queue_item",
+        notify,
+    )
+    engine = ReleaseManagerEngine(db_session, assessor=_assessor(_report()))
+    task = await engine.run_cycle()
+    assert task is not None
+    notify.assert_awaited_once()
+    _args, kwargs = notify.await_args
+    assert kwargs["kind"] == "release"
+    assert kwargs["id8"] == str(task.id)[:8]
+    assert _VERSION in kwargs["title"]
+
+
+@pytest.mark.asyncio
+async def test_proposes_survives_telegram_push_failure(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A Telegram send failure must never block origination itself."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    monkeypatch.setattr(
+        "roboco.services.notification_delivery.NotificationDeliveryService."
+        "notify_ceo_of_queue_item",
+        AsyncMock(side_effect=RuntimeError("boom")),
+    )
+    engine = ReleaseManagerEngine(db_session, assessor=_assessor(_report()))
+    task = await engine.run_cycle()
+    assert task is not None
+    assert await get_task_service(db_session).list_open_release_proposals()
+
+
 @pytest.mark.asyncio
 async def test_none_assessment_no_proposal(
     db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch

tests/unit/services/test_telegram_client.py

@@ -93,6 +93,59 @@ async def test_live_client_send_message_with_reply_markup_and_message_id() -> No
     await client.close()
 
 
+@pytest.mark.asyncio
+async def test_live_client_send_message_with_formatting_fields() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content.decode())
+        assert body["parse_mode"] == "HTML"
+        assert body["link_preview_options"] == {"is_disabled": True}
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 1}})
+
+    transport = httpx.MockTransport(handler)
+    http_client = httpx.AsyncClient(transport=transport)
+    client = LiveTelegramClient(_CREDS, timeout=5.0, client=http_client)
+    result = await client.send_message(
+        "<b>hi</b>", parse_mode="HTML", disable_link_preview=True
+    )
+    assert result.sent is True
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_live_client_send_message_omits_formatting_fields_when_unset() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content.decode())
+        assert "parse_mode" not in body
+        assert "link_preview_options" not in body
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 1}})
+
+    transport = httpx.MockTransport(handler)
+    http_client = httpx.AsyncClient(transport=transport)
+    client = LiveTelegramClient(_CREDS, timeout=5.0, client=http_client)
+    result = await client.send_message("hi")
+    assert result.sent is True
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_live_client_edit_message_text_with_parse_mode() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/bot123456:ABC/editMessageText"
+        body = json.loads(request.content.decode())
+        assert body["text"] == "<b>done</b>"
+        assert body["parse_mode"] == "HTML"
+        assert body["link_preview_options"] == {"is_disabled": True}
+        return httpx.Response(200, json={"ok": True, "result": True})
+
+    transport = httpx.MockTransport(handler)
+    http_client = httpx.AsyncClient(transport=transport)
+    client = LiveTelegramClient(_CREDS, timeout=5.0, client=http_client)
+    await client.edit_message_text(
+        7, "<b>done</b>", parse_mode="HTML", disable_link_preview=True
+    )
+    await client.close()
+
+
 @pytest.mark.asyncio
 async def test_live_client_get_updates_success() -> None:
     def handler(request: httpx.Request) -> httpx.Response:
@@ -161,4 +214,5 @@ async def test_null_client_v2_methods_are_all_noops() -> None:
     # None of these raise — that's the whole contract.
     await client.answer_callback_query("cq1", "text")
     await client.edit_message_reply_markup(1, None)
-    await client.edit_message_text(1, "done")
+    await client.edit_message_text(1, "done", parse_mode="HTML")
+    await client.send_message("hi", parse_mode="HTML", disable_link_preview=True)

tests/unit/services/test_telegram_inbound_engine.py

@@ -253,7 +253,7 @@ async def test_expired_reply_prompt_sends_notice(
     )
 
     client.send_message.assert_awaited_once_with(
-        "That prompt expired — tap the button again."
+        "That prompt expired — tap the button again.", parse_mode="HTML"
     )
     dispatch.assert_not_called()
     assert ("777", 42) not in ti._PENDING_REPLIES
@@ -743,3 +743,207 @@ async def test_mark_audit_adds_via_telegram_row() -> None:
     assert added.agent_id == CEO_UUID
     assert added.target_id == task_id
     assert added.details["via"] == "telegram"
+
+
+# ---------------------------------------------------------------------------
+# /queue rendering — pluralization + HTML formatting
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_send_queue_empty_says_nothing_awaiting(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = _engine()
+    monkeypatch.setattr(engine, "_collect_queue_items", AsyncMock(return_value=[]))
+    client = AsyncMock()
+
+    await engine._send_queue(client)
+
+    client.send_message.assert_awaited_once_with(
+        "✅ Nothing awaiting your approval.", parse_mode="HTML"
+    )
+
+
+@pytest.mark.asyncio
+async def test_send_queue_singular_item_pluralization(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = _engine()
+    monkeypatch.setattr(
+        engine,
+        "_collect_queue_items",
+        AsyncMock(return_value=[("task", "a1b2c3d4", "", "Ship it")]),
+    )
+    client = AsyncMock()
+
+    await engine._send_queue(client)
+
+    header = client.send_message.await_args_list[0].args[0]
+    assert header == "<b>🔔 Awaiting your approval</b> — 1 item"
+
+
+@pytest.mark.asyncio
+async def test_send_queue_plural_items_pluralization(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = _engine()
+    monkeypatch.setattr(
+        engine,
+        "_collect_queue_items",
+        AsyncMock(
+            return_value=[
+                ("task", "a1b2c3d4", "", "Ship it"),
+                ("release", "deadbeef", "", "v1.0.0 ready"),
+            ]
+        ),
+    )
+    client = AsyncMock()
+
+    await engine._send_queue(client)
+
+    header = client.send_message.await_args_list[0].args[0]
+    assert header == "<b>🔔 Awaiting your approval</b> — 2 items"
+
+
+@pytest.mark.asyncio
+async def test_send_queue_item_line_escapes_title_and_carries_keyboard(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Injection regression: a malicious task title must arrive HTML-escaped
+    — never as live markup — in the /queue item line's sent payload."""
+    engine = _engine()
+    monkeypatch.setattr(
+        engine,
+        "_collect_queue_items",
+        AsyncMock(return_value=[("task", "a1b2c3d4", "", "<b>bold&joke</b>")]),
+    )
+    client = AsyncMock()
+
+    await engine._send_queue(client)
+
+    item_call = client.send_message.await_args_list[1]
+    text = item_call.args[0]
+    assert "&lt;b&gt;bold&amp;joke&lt;/b&gt;" in text
+    assert "<b>bold&joke</b>" not in text
+    assert text.startswith("📋 <b>Task</b> — ")
+    assert item_call.kwargs["parse_mode"] == "HTML"
+    assert "reply_markup" in item_call.kwargs
+
+
+# ---------------------------------------------------------------------------
+# /task — link preview disabled, title/status/team escaping
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_dispatch_command_task_disables_link_preview(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = _engine()
+    monkeypatch.setattr(engine, "_render_task", AsyncMock(return_value="detail"))
+    client = AsyncMock()
+
+    await engine._dispatch_command("task", "a1b2c3d4", client)
+
+    client.send_message.assert_awaited_once_with(
+        "detail", parse_mode="HTML", disable_link_preview=True
+    )
+
+
+def test_format_task_detail_escapes_html_in_title() -> None:
+    """Injection regression: a task titled ``<b>bold&joke</b>`` must render
+    HTML-escaped, not as live markup, in /task's detail view."""
+    engine = _engine()
+    task = _fake_task(title="<b>bold&joke</b>")
+
+    rendered = engine._format_task_detail(task)
+
+    assert "&lt;b&gt;bold&amp;joke&lt;/b&gt;" in rendered
+    assert "<b>bold&joke</b>" not in rendered
+
+
+def test_format_task_detail_pr_url_is_a_named_link() -> None:
+    engine = _engine()
+    task = _fake_task()
+    task.pr_url = "https://github.com/example/repo/pull/1"
+
+    rendered = engine._format_task_detail(task)
+
+    assert '<a href="https://github.com/example/repo/pull/1">View PR</a>' in rendered
+
+
+def test_format_task_detail_pr_url_quote_cannot_break_out_of_href() -> None:
+    """Injection regression: a pr_url containing a literal '"' must not be
+    able to close the href attribute early and inject a bogus attribute —
+    _esc_attr (quote=True) turns it into '&quot;', keeping the whole value
+    inside the attribute."""
+    engine = _engine()
+    task = _fake_task()
+    task.pr_url = 'https://evil.example/x" onmouseover="alert(1)'
+
+    rendered = engine._format_task_detail(task)
+
+    assert (
+        '<a href="https://evil.example/x&quot; onmouseover=&quot;alert(1)">'
+        "View PR</a>" in rendered
+    )
+    assert 'onmouseover="alert(1)"' not in rendered
+
+
+# ---------------------------------------------------------------------------
+# outcome confirmations (_finish_action / _consume_reply) — escaping
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_finish_action_escapes_text_and_edits_origin() -> None:
+    engine = _engine()
+    client = AsyncMock()
+
+    await engine._finish_action(client, 42, True, "Rejected: <script>xss</script>")
+
+    client.edit_message_reply_markup.assert_awaited_once_with(42, None)
+    call = client.edit_message_text.await_args
+    assert call.args == (42, "✅ Rejected: &lt;script&gt;xss&lt;/script&gt;")
+    assert call.kwargs["parse_mode"] == "HTML"
+
+
+@pytest.mark.asyncio
+async def test_finish_action_escapes_text_without_origin() -> None:
+    engine = _engine()
+    client = AsyncMock()
+
+    await engine._finish_action(client, None, False, "<script>alert(1)</script>")
+
+    call = client.send_message.await_args
+    assert call.args == ("❌ &lt;script&gt;alert(1)&lt;/script&gt;",)
+    assert call.kwargs["parse_mode"] == "HTML"
+
+
+@pytest.mark.asyncio
+async def test_consume_reply_reject_outcome_arrives_escaped(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """End-to-end regression: a reject reason that reaches the CEO through
+    whatever text a dispatch handler returns must never arrive as live HTML —
+    the funnel (_finish_action) escapes it regardless of the handler."""
+    engine = _engine()
+    client = AsyncMock()
+    dispatch = AsyncMock(return_value=(True, "Rejected: <script>xss</script>"))
+    monkeypatch.setattr(engine, "_dispatch_reject", dispatch)
+    pending = ti._PendingAction(
+        kind="xpost",
+        id8="a1b2c3d4",
+        extra="",
+        action="reject",
+        origin_message_id=None,
+        expires_at=time.monotonic() + 60,
+    )
+
+    await engine._consume_reply(pending, "<script>xss</script>", client)
+
+    dispatch.assert_awaited_once_with("xpost", "a1b2c3d4", "", "<script>xss</script>")
+    sent_text = client.send_message.await_args.args[0]
+    assert "&lt;script&gt;xss&lt;/script&gt;" in sent_text
+    assert "<script>xss</script>" not in sent_text

tests/unit/services/test_telegram_inbound_router.py

@@ -6,12 +6,17 @@ from __future__ import annotations
 import pytest
 from roboco.config import settings as cfg
 from roboco.services.telegram_inbound import (
+    _MESSAGE_CHAR_LIMIT,
     ParsedCallback,
     _authorized_chat,
+    _esc,
+    _esc_attr,
+    _truncate,
     build_action_keyboard,
     build_callback,
     parse_callback,
     parse_command,
+    render_queue_item_text,
 )
 
 _CALLBACK_DATA_MAX_BYTES = 64  # mirrors Telegram's own callback_data cap
@@ -134,3 +139,94 @@ class TestAuthorizedChat:
 
     def test_empty_chat_id_rejected(self) -> None:
         assert _authorized_chat("", "12345") is False
+
+
+class TestEsc:
+    def test_escapes_angle_brackets_and_ampersand(self) -> None:
+        assert _esc("<b>bold&joke</b>") == "&lt;b&gt;bold&amp;joke&lt;/b&gt;"
+
+    def test_quotes_are_left_alone(self) -> None:
+        # quote=False — _esc renders HTML text nodes, where quotes need no
+        # escaping. Attribute values (e.g. href="...") go through _esc_attr
+        # instead, which does escape them.
+        assert _esc('it\'s "fine"') == 'it\'s "fine"'
+
+    def test_stringifies_non_str_values(self) -> None:
+        assert _esc(42) == "42"
+
+
+class TestEscAttr:
+    def test_escapes_quotes_and_angle_brackets(self) -> None:
+        assert _esc_attr("""a"b'c<d>e""") == "a&quot;b&#x27;c&lt;d&gt;e"
+
+    def test_stringifies_non_str_values(self) -> None:
+        assert _esc_attr(42) == "42"
+
+
+class TestTruncateHtml:
+    def test_short_text_is_untouched(self) -> None:
+        assert _truncate("hello") == "hello"
+
+    def test_backs_off_before_an_unclosed_angle_bracket(self) -> None:
+        # A naive slice at `limit - 1` would land inside "<code>", leaving a
+        # bare '<' Telegram's HTML parser can't make sense of — back off to
+        # before it instead.
+        text = ("x" * 4093) + "<code>"
+        result = _truncate(text)
+        assert result.endswith("…")
+        assert not result.rstrip("…").endswith("<")
+
+    def test_backs_off_before_an_unclosed_entity(self) -> None:
+        text = ("x" * 4093) + "&amp;"
+        result = _truncate(text)
+        assert result.endswith("…")
+        assert "&am" not in result
+
+    @pytest.mark.parametrize("title_len", range(4048, 4069))
+    def test_render_queue_item_truncation_balances_code_tag(
+        self, title_len: int
+    ) -> None:
+        # Regression: a naive char-count slice landed inside the trailing
+        # `<code>id8</code>` span, shipping an unclosed `<code>` Telegram's
+        # HTML parser rejects outright.
+        text = render_queue_item_text("roadmap", "abc12345", "item-0", "A" * title_len)
+        assert text.count("<code>") == text.count("</code>")
+        assert len(text) <= _MESSAGE_CHAR_LIMIT
+
+    def test_truncate_balances_a_bold_wrapped_tag(self) -> None:
+        text = "<b>" + ("y" * 4200) + "</b>"
+        result = _truncate(text)
+        assert result.count("<b>") == result.count("</b>")
+        assert len(result) <= _MESSAGE_CHAR_LIMIT
+
+
+class TestRenderQueueItemText:
+    def test_escapes_html_in_title(self) -> None:
+        text = render_queue_item_text("task", "a1b2c3d4", "", "<b>bold&joke</b>")
+        assert "&lt;b&gt;bold&amp;joke&lt;/b&gt;" in text
+        assert "<b>bold&joke</b>" not in text
+
+    def test_kind_emoji_and_label_per_kind(self) -> None:
+        assert render_queue_item_text("release", "a1b2c3d4", "", "x").startswith(
+            "🚀 <b>Release</b>"
+        )
+        assert render_queue_item_text("video", "a1b2c3d4", "", "x").startswith(
+            "🎬 <b>Video</b>"
+        )
+        assert render_queue_item_text("xpost", "a1b2c3d4", "", "x").startswith(
+            "✕ <b>Post</b>"
+        )
+        assert render_queue_item_text("roadmap", "a1b2c3d4", "", "x").startswith(
+            "🗺️ <b>Roadmap</b>"
+        )
+        assert render_queue_item_text("task", "a1b2c3d4", "", "x").startswith(
+            "📋 <b>Task</b>"
+        )
+
+    def test_id8_and_extra_render_as_code_span(self) -> None:
+        text = render_queue_item_text("roadmap", "a1b2c3d4", "item-2", "x")
+        assert "<code>a1b2c3d4:item-2</code>" in text
+
+    def test_no_extra_omits_suffix(self) -> None:
+        text = render_queue_item_text("task", "a1b2c3d4", "", "x")
+        assert "<code>a1b2c3d4</code>" in text

tests/unit/services/test_video_engine.py

@@ -456,6 +456,68 @@ async def test_originate_video_post_holds_draft_for_secretary(
     assert draft["source_task_id"] == str(source_task.id)  # traceability back-ref
 
 
+@pytest.mark.asyncio
+async def test_originate_video_post_sends_telegram_push(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed(db_session)
+    _enable(monkeypatch)
+    notify = AsyncMock()
+    monkeypatch.setattr(
+        "roboco.services.notification_delivery.NotificationDeliveryService."
+        "notify_ceo_of_queue_item",
+        notify,
+    )
+    engine = video_engine_module.VideoEngine(db_session)
+    source_task = await engine.open_video_task(
+        occasion="release v1.0.0",
+        script="Here's what shipped",
+        platforms=["x", "tiktok"],
+        brief="Announce the release",
+    )
+    assert source_task is not None
+
+    draft_task = await engine._originate_video_post(
+        source_task=source_task,
+        mp4_paths={"vertical": "/a.mp4", "square": "/b.mp4"},
+        captions={"x": "shipped!", "tiktok": "shipped!"},
+        platforms=["x", "tiktok"],
+    )
+
+    notify.assert_awaited_once()
+    _args, kwargs = notify.await_args
+    assert kwargs["kind"] == "video"
+    assert kwargs["id8"] == str(draft_task.id)[:8]
+    assert kwargs["title"] == "release v1.0.0"
+
+
+@pytest.mark.asyncio
+async def test_originate_video_post_survives_telegram_push_failure(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A Telegram send failure must never block the draft itself."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    monkeypatch.setattr(
+        "roboco.services.notification_delivery.NotificationDeliveryService."
+        "notify_ceo_of_queue_item",
+        AsyncMock(side_effect=RuntimeError("boom")),
+    )
+    engine = video_engine_module.VideoEngine(db_session)
+    source_task = await engine.open_video_task(
+        occasion="release v1.0.0", script="s", platforms=["x"], brief="b"
+    )
+    assert source_task is not None
+
+    draft_task = await engine._originate_video_post(
+        source_task=source_task,
+        mp4_paths={"vertical": "/a.mp4", "square": "/b.mp4"},
+        captions={"x": "shipped!"},
+        platforms=["x"],
+    )
+    assert draft_task is not None
+
+
 @pytest.mark.asyncio
 async def test_originate_video_post_not_counted_by_dedupe_against_new_occasion(
     db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch

tests/unit/services/test_x_engine.py

@@ -252,6 +252,54 @@ async def test_draft_release_post_dedupes_same_version(
     assert len(open_posts) == ONE
 
 
+@pytest.mark.asyncio
+async def test_originate_post_sends_telegram_push(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """``_originate_post`` is the shared chokepoint for all three X sources
+    (release/reply/feature) — a freshly-drafted post fires the styled push
+    DM (xpost kind, the draft's id8, its body)."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    _mock_local_model(monkeypatch, "RoboCo just shipped a great new feature!")
+    notify = AsyncMock()
+    monkeypatch.setattr(
+        "roboco.services.notification_delivery.NotificationDeliveryService."
+        "notify_ceo_of_queue_item",
+        notify,
+    )
+    engine = x_engine_module.XEngine(db_session, client=_FakeClient())
+    task = await engine.draft_release_post(
+        version=_VERSION, highlights=["feat: new thing"]
+    )
+    assert task is not None
+    notify.assert_awaited_once()
+    _args, kwargs = notify.await_args
+    assert kwargs["kind"] == "xpost"
+    assert kwargs["id8"] == str(task.id)[:8]
+    body = markers.get_x_draft_body(task)
+    assert body is not None
+    assert kwargs["title"] == body[:100]
+
+
+@pytest.mark.asyncio
+async def test_originate_post_survives_telegram_push_failure(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A Telegram send failure must never block the draft itself."""
+    await _seed(db_session)
+    _enable(monkeypatch)
+    _mock_local_model(monkeypatch, "shipped!")
+    monkeypatch.setattr(
+        "roboco.services.notification_delivery.NotificationDeliveryService."
+        "notify_ceo_of_queue_item",
+        AsyncMock(side_effect=RuntimeError("boom")),
+    )
+    engine = x_engine_module.XEngine(db_session, client=_FakeClient())
+    task = await engine.draft_release_post(version=_VERSION, highlights=[])
+    assert task is not None
+
+
 @pytest.mark.asyncio
 async def test_draft_release_post_respects_open_cap(
     db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch