feat(branding): channel-only hop gate, link/@ replace maps, prompt library, branding failure trail

README.md

@@ -198,6 +198,17 @@ AutoTelegramForward v2.3 is engineered with strict, fail-closed microservice bou
 
 ## 🚀 What is New in v2.x
 
+- 📡 **Branding Hop with a Channel-Only Gate (A → C → B)**:
+  Route channel content through your own branding channel so the target's header names it — while messages written by normal users are forwarded directly with their own name (`branding_require_channel`). Configurable from the bot's smart menu, the dashboard's rule editor and the mobile client; the trigger policy (origins, keywords, match mode, forward/channel gates) is one shared decision across the Python engine, the Go worker and the preview, pinned by cross-engine parity fixtures.
+- 🔁 **Replace Any Link Found**:
+  `link_replace_enabled` + a rewrite map with three key shapes: `*` rewrites every link to one target, a domain key rewrites one domain, a URL-prefix key rewrites the deep links it starts with. Applied entity-aware by the link engine in both engines.
+- ❕ **Replace Any @ID**:
+  `mention_rewrite_map` rewrites handles (`@oldshop → @newshop`, or `* → @main` for every mention) in the final replacement pass — after the AI stage, so a rewrite cannot resurrect a replaced handle. Dedicated editor panel in the bot (smart menu → 🔁 Link & @replace maps), registry rows on the dashboard and the mobile settings sheet.
+- 🧠 **Prompt Library Management (bot + dashboard + mobile)**:
+  The global prompt manager (`AI → 🧠 Prompts`) closes the bot's missing surface: list, create, rename, edit the system prompt / user template / description, test against a real provider, and **enable/disable** — a disabled template stays in the library while deliveries fall back to the task's AI-config prompt (`[PIPELINE:AI_REWRITE:PROMPT_OFF]`). The mobile client gains the same library (CRUD + test) over the shared `/api/prompts` contract.
+- 🚨 **Branding Failure Trail**:
+  Hop failures now carry their own event codes (`BRANDING_HOP_FAILED`, `BRANDING_STAGE1_FAILED`, `BRANDING_STAGE2_FAILED`) into the Rust log store and the Log Center, so "the branded route is broken for this task" is one filter instead of a needle in every delivery failure.
+
 - 🧹 **Advanced Content Cleaner & Sanitizer**:
   Strip links, `@mentions`, `#hashtags`, emojis, and custom regex patterns per rule, with preview and dynamic toggle.
 - 🧬 **Message Deduplication Engine**:

core/application/backup_service.py

@@ -299,6 +299,9 @@ def _prompt_dict(self, t: PromptTemplate) -> Dict[str, Any]:
             "current_version": t.current_version,
             "is_system": t.is_system,
             "owner_user_id": int(getattr(t, "owner_user_id", 0) or 0),
+            # The library enable/disable switch rides the backup so a restore
+            # brings back the same active set of prompts.
+            "enabled": bool(getattr(t, "enabled", True)),
         }
 
     # ------------------------------------------------------------------ import
@@ -521,6 +524,7 @@ async def restore_snapshot(
                         if target_owner_user_id is not None
                         else raw.get("owner_user_id") or 0
                     ),
+                    enabled=bool(raw.get("enabled", True)),
                 )
                 await self._prompts.add_template(t)
                 p_stats["imported"] += 1

core/application/pipeline.py

@@ -517,6 +517,16 @@ async def _perform_brand_hop(self, ctx: "PipelineContext", rule: ForwardRule,
                 "[PIPELINE:BRANDING] [%s] Rule %s hop failed (%s) — delivering directly",
                 ctx.correlation_id[:8], rule.id[:8], exc,
             )
+            # Structured error trail: the branded hop is a delivery-stage side
+            # effect, so its failure is an audit event with its own reason code
+            # (the Rust log store and the Log Center filter on it) — not just a
+            # stdout line the operator has to grep for.
+            await self._audit(
+                ctx, "forward.branding_failed", "FAILED", severity="ERROR",
+                reason_code="BRANDING_HOP_FAILED", rule_id=rule.id,
+                message_id=str(getattr(ctx.payload, "message_id", "") or ""),
+                result_summary=f"branding hop failed: {type(exc).__name__}",
+            )
             return None
         if plan:
             logger.info(
@@ -530,6 +540,12 @@ async def _perform_brand_hop(self, ctx: "PipelineContext", rule: ForwardRule,
                 "[PIPELINE:BRANDING] [%s] Rule %s msg %s: nothing posted into C — delivering directly",
                 ctx.correlation_id[:8], rule.id[:8], getattr(ctx.payload, "message_id", "?"),
             )
+            await self._audit(
+                ctx, "forward.branding_failed", "FAILED", severity="ERROR",
+                reason_code="BRANDING_HOP_EMPTY", rule_id=rule.id,
+                message_id=str(getattr(ctx.payload, "message_id", "") or ""),
+                result_summary="branding hop posted nothing into C",
+            )
         return plan
 
     # ----------------------------------------------------------------- #
@@ -1526,7 +1542,17 @@ async def _apply_transforms(
                 )
 
         # Link Management Engine (Entity-aware, domain allowlist/blocklist, SSRF protected)
+        # "Replace any link found": with `link_replace_enabled` on and a
+        # rewrite map configured, a task with no stricter policy upgrades to
+        # REWRITE_LINKS so the engine rewrites the links the map names (a `*`
+        # key rewrites every link) instead of leaving them untouched.
         link_policy = getattr(rule, "link_policy", "PRESERVE_ALL")
+        if (
+            bool(getattr(rule, "link_replace_enabled", False))
+            and (getattr(rule, "link_rewrite_map", None) or {})
+            and str(link_policy or "").strip().upper() in ("", "PRESERVE_ALL")
+        ):
+            link_policy = "REWRITE_LINKS"
         if (link_policy and link_policy != "PRESERVE_ALL") or getattr(rule, "remove_links", False):
             clean_text, updated_ents, links_mod = LinkManagementEngine.apply_to_payload(
                 payload, rule, text=text
@@ -1634,6 +1660,22 @@ async def _apply_transforms(
         if getattr(rule, "replacements", None):
             text = self._filter_engine.replace_text(text, rule.replacements)
 
+        # "@mention rewrites" run in the same final pass: a map-driven replace
+        # of @handles (`@oldshop -> @newshop`, or a `*` catch-all), so a
+        # rewrite cannot reintroduce a handle the task asked to replace.
+        mention_map = getattr(rule, "mention_rewrite_map", None) or {}
+        if mention_map:
+            from ..domain.text_replace import apply_mention_rewrites
+
+            rewritten = apply_mention_rewrites(text, mention_map)
+            if rewritten != text:
+                logger.info(
+                    "[PIPELINE:REPLACE] [%s] Rule %s rewrote @mentions of msg %s (len %d -> %d)",
+                    ctx.correlation_id[:8], rule.id[:8], payload.message_id,
+                    len(text), len(rewritten),
+                )
+                text = rewritten
+
         # Header & Footer
         header = getattr(rule, "header", "") or getattr(rule, "header_text", "")
         footer = getattr(rule, "footer", "") or getattr(rule, "footer_text", "")
@@ -1742,6 +1784,24 @@ async def _run_ai_transform(
         if isinstance(rule.metadata, dict) and rule.metadata.get("prompt_id"):
             prompt_id = str(rule.metadata.get("prompt_id"))
 
+        if self._prompt_service and prompt_id:
+            # A template switched off in the prompt manager stays in the
+            # library but stops steering deliveries: the task is rewritten
+            # with its AI config's own prompt, exactly as if none were set.
+            # Preview and the tester keep working on disabled templates —
+            # that is how an operator checks a prompt before re-enabling it.
+            try:
+                template = await self._prompt_service.get_template(prompt_id)
+            except Exception:
+                template = None
+            if template is not None and not bool(getattr(template, "enabled", True)):
+                logger.info(
+                    "[PIPELINE:AI_REWRITE:PROMPT_OFF] [%s] Rule %s prompt '%s' is disabled — "
+                    "using the AI config's own prompt",
+                    ctx.correlation_id[:8], rule.id[:8], prompt_id,
+                )
+                prompt_id = ""
+                prompt_ver = 0
         if self._prompt_service and prompt_id:
             try:
                 preview = await self._prompt_service.preview_prompt(

core/application/prompt_service.py

@@ -349,6 +349,23 @@ async def activate_version(self, template_id: str, version: int) -> bool:
         """Activates a specific version."""
         return await self.prompt_repo.activate_version(template_id, version)
 
+    async def set_enabled(self, template_id: str, enabled: bool) -> PromptTemplate:
+        """Enable or disable a template without touching its version history.
+
+        Disabling is a delivery-policy switch, not a content change: the
+        pipeline stops consulting the template (the task falls back to its AI
+        config's own prompt) but the library row, its versions and its
+        assignment on tasks stay exactly as they are, so re-enabling restores
+        the previous behavior with nothing re-entered.
+        """
+        template = await self.prompt_repo.get_template(template_id)
+        if not template:
+            raise ValueError(f"Prompt template '{template_id}' not found.")
+        template.enabled = bool(enabled)
+        template.updated_at = int(time.time())
+        await self.prompt_repo.update_template(template)
+        return template
+
     async def get_template(self, template_id: str) -> Optional[PromptTemplate]:
         return await self.prompt_repo.get_template(template_id)
 

core/domain/branding.py

@@ -26,6 +26,13 @@
 * ``match_mode`` picks ``any`` (at least one group matches) or ``all``.
 * ``require_forward`` is a *gate*, not a check: it refuses a message that is not
   itself a forward, regardless of match_mode.
+* ``require_channel`` is the same kind of gate for the *sender side*: only a
+  channel-originated message earns the hop (a native channel post in A, or a
+  forward whose origin is a channel). A message written by a normal user fails
+  the gate and is delivered directly — which is what keeps the user's own name
+  in B's "Forwarded from" header while channel content gets branded. The two
+  gates compose, so ``require_forward + require_channel`` reads "only content
+  that arrived as a channel forward".
 * ``force_forward`` (default true) is what makes stage 2 a native forward from
   C, which is the only way B can show C in its header.
 """
@@ -43,6 +50,7 @@
     "channel_id",
     "enabled",
     "force_forward",
+    "is_channel_origin",
     "match_origin",
     "match_keywords",
     "should_hop",
@@ -63,11 +71,12 @@
     "keywords": [],
     "match_mode": "any",
     "require_forward": False,
+    "require_channel": False,
     "force_forward": True,
 }
 
 _LIST_KEYS = ("origin_ids", "origin_titles", "keywords")
-_BOOL_KEYS = ("triggers_enabled", "require_forward", "force_forward")
+_BOOL_KEYS = ("triggers_enabled", "require_forward", "require_channel", "force_forward")
 
 
 # --------------------------------------------------------------------------- #
@@ -174,6 +183,30 @@ def _origin_of(payload: Any) -> Dict[str, Any]:
     return origin if isinstance(origin, dict) else {}
 
 
+def is_channel_origin(payload: Any) -> bool:
+    """True when the message *comes from a channel*.
+
+    Two shapes count, because Telegram produces both:
+
+    * the message was **forwarded from a channel** — the forward-origin header
+      names a channel (the origin ``type`` carries "channel"; the pyrogram
+      adapter spells it ``channel``/``MessageForwardOriginChannel`` depending
+      on the client path, so the test is a substring, never an equality);
+    * or the message is a **native channel post in A** — the source chat A is
+      itself a channel (``chat_type == "channel"``), so the sender is the
+      channel rather than a person.
+
+    A message written by a normal user in a group (or forwarded from a user)
+    answers False — that is the point: the ``require_channel`` gate lets user
+    messages keep their own "Forwarded from <user>" header on a direct
+    delivery while channel content takes the branded hop.
+    """
+    origin = _origin_of(payload)
+    if origin and "channel" in str(origin.get("type", "")).lower():
+        return True
+    return str(getattr(payload, "chat_type", "") or "").strip().lower() == "channel"
+
+
 def match_origin(payload: Any, origin_ids: List[str], origin_titles: List[str]) -> bool:
     """Match the *forward origin channel* (Telegram's "Forwarded from" header).
 
@@ -254,6 +287,8 @@ def should_hop(
     # A "require" is a gate and must hold on its own, whatever match_mode says.
     if cfg["require_forward"] and not _origin_of(payload):
         return False, "branding_requires_forward_origin"
+    if cfg["require_channel"] and not is_channel_origin(payload):
+        return False, "branding_requires_channel_origin"
 
     conditions: List[str] = []
     matched: List[str] = []

core/domain/entities.py

@@ -1346,6 +1346,22 @@ def branding_require_forward(self) -> bool:
     def branding_require_forward(self, val: Any) -> None:
         self._set_branding_key("require_forward", bool(val))
 
+    @property
+    def branding_require_channel(self) -> bool:
+        """Gate: refuse the hop unless the message originates from a channel.
+
+        This is the switch that makes the flow "normal users are forwarded
+        directly with their own name, channel content goes through C": a
+        user's own post fails the gate and keeps its personal "Forwarded
+        from <user>" header on a direct delivery, while a channel post (or a
+        channel forward) earns the branded A -> C -> B route.
+        """
+        return bool(self._branding_key("require_channel"))
+
+    @branding_require_channel.setter
+    def branding_require_channel(self, val: Any) -> None:
+        self._set_branding_key("require_channel", bool(val))
+
     @property
     def branding_force_forward(self) -> bool:
         """Stage 2 sends a native forward of C — the only way B can show C."""
@@ -1675,6 +1691,41 @@ def smart_ai_rewrite(self, val: bool) -> None:
             self.metadata = {}
         self.metadata["smart_ai_rewrite"] = bool(val)
 
+    # --- Replace engine: link & mention rewrite maps -----------------------
+    # "Replace any link found" and "replace any @ID" are the two map-driven
+    # replacement switches. The maps ride the rule's metadata blob so they
+    # need no schema migration, and the registry exposes them (with the
+    # column-backed `link_rewrite_map`) so the bot, dashboard, mobile client
+    # and the REST settings surface all edit the same values.
+    @property
+    def link_replace_enabled(self) -> bool:
+        """Rewrite every link the text carries through ``link_rewrite_map``."""
+        return bool(self.metadata.get("link_replace_enabled", False)) if isinstance(self.metadata, dict) else False
+
+    @link_replace_enabled.setter
+    def link_replace_enabled(self, val: bool) -> None:
+        if not isinstance(self.metadata, dict):
+            self.metadata = {}
+        self.metadata["link_replace_enabled"] = bool(val)
+
+    @property
+    def mention_rewrite_map(self) -> Dict[str, str]:
+        """@username -> @username pairs applied to the message text.
+
+        A ``*`` key replaces every mention with its value; otherwise only the
+        exact (case-insensitive) handles listed are rewritten.
+        """
+        if not isinstance(self.metadata, dict):
+            return {}
+        raw = self.metadata.get("mention_rewrite_map", {})
+        return dict(raw) if isinstance(raw, dict) else {}
+
+    @mention_rewrite_map.setter
+    def mention_rewrite_map(self, val: Any) -> None:
+        if not isinstance(self.metadata, dict):
+            self.metadata = {}
+        self.metadata["mention_rewrite_map"] = dict(val or {})
+
     @property
     def ai_rewrite_active(self) -> bool:
         """Single authority for "does this task rewrite text with an LLM?".
@@ -2465,6 +2516,10 @@ class PromptTemplate:
     updated_at: int = field(default_factory=_now)
     # v12: user prompts are per-tenant; system prompts (is_system) stay shared.
     owner_user_id: int = 0
+    # A disabled template stays in the library (editable, testable) but the
+    # delivery pipeline skips it: a task pointing at it is rewritten with the
+    # AI config's own prompt instead, exactly as if no template were set.
+    enabled: bool = True
 
 
 @dataclass

core/domain/rule_settings.py

@@ -108,6 +108,27 @@ def _s(**kw: Any) -> SettingSpec:
                            "(`i` ignore case, `c` case-sensitive, `w` whole word)."
                        )),
     "link_replacement": _s(kind="str", default="", group="cleanup"),
+    # --- Replace engine: links & @mentions ---------------------------------
+    # The toggle arms the map below it: with `link_replace_enabled` on, every
+    # link the text carries is rewritten through `link_rewrite_map` (a `*`
+    # key rewrites every link to one target, a domain key rewrites that
+    # domain, a key containing "/" rewrites the URLs it prefixes).
+    "link_replace_enabled": _s(
+        kind="bool", default=False, group="cleanup",
+        help="Rewrite every link found in the text through the link rewrite map."),
+    # Column-backed like allowed_media_types: the Link Management Engine reads
+    # the real forward_rules column, and backups/the REST contract already
+    # carry it. Registered so the API settings surface and the generic
+    # settings UIs can edit the map without a bespoke route.
+    "link_rewrite_map": _s(
+        kind="dict", default={}, group="cleanup",
+        help='Link rewrite pairs: {"*": "https://t.me/me"} rewrites every link,'
+             ' "olddomain.com": "newdomain.com" rewrites one domain,'
+             ' "https://old.t.me/x": "https://new.t.me/y" rewrites a full URL.'),
+    "mention_rewrite_map": _s(
+        kind="dict", default={}, group="cleanup",
+        help='@mention rewrite pairs: {"@oldshop": "@newshop"} rewrites one'
+             ' handle (case-insensitive), "*": "@main" rewrites every @ID.'),
     "clean_usernames": _s(kind="bool", default=False, group="cleanup"),
     "clean_hashtags": _s(kind="bool", default=False, group="cleanup"),
     "clean_phones": _s(kind="bool", default=False, group="cleanup"),
@@ -303,6 +324,9 @@ def _s(**kw: Any) -> SettingSpec:
     "branding_require_forward": _s(
         kind="bool", default=False, group="branding",
         help="Refuse the hop unless the message is itself a forward."),
+    "branding_require_channel": _s(
+        kind="bool", default=False, group="branding",
+        help="Refuse the hop unless the message comes from a channel — normal users are then forwarded directly with their own name."),
     "branding_force_forward": _s(
         kind="bool", default=True, group="branding",
         help="Send the final step as a native forward of C, so B's header names C."),