feat(whatsapp): tag owner-typed inbound text with [owner reply] prefix

When WHATSAPP_FORWARD_OWNER_MESSAGES is enabled and the bridge marks an
inbound message with fromOwner=true, also prefix MessageEvent.text with
"[owner reply] " at construction time. This makes the disambiguation
survive any downstream plugin failure (e.g. handover-rule errors that
bypass silent_ingest), so transcripts never misattribute owner-typed
text to the customer.

Idempotent: re-applies are guarded so a future producer that pre-tags
text won't be double-prefixed.

plugins/platforms/whatsapp/adapter.py

@@ -35,6 +35,10 @@ from hermes_constants import (
 
 logger = logging.getLogger(__name__)
 
+# Inbound owner-typed WhatsApp text is prefixed at MessageEvent construction so
+# transcripts stay disambiguated even if downstream plugins fail before silent_ingest.
+_OWNER_REPLY_PREFIX = "[owner reply] "
+
 
 def _listener_pids_on_port(port: int) -> list:
     """PIDs of processes *listening* on ``port`` (POSIX) — never clients.
@@ -1314,12 +1318,16 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
             # that look owner-typed (linked-device send, not echoed from our
             # own /send).  Surfaced under a platform-prefixed key so plugins
             # can detect "owner just replied in this customer chat" without
-            # having to peek at raw_message.  Gated by
-            # ``WHATSAPP_FORWARD_OWNER_MESSAGES`` at the bridge layer; the
-            # propagation here is unconditional so a future producer can set
-            # the flag without us having to touch this code path again.
+            # having to peek at raw_message.  We also prefix ``MessageEvent.text``
+            # with ``[owner reply] `` here so the marker survives any downstream
+            # failure (e.g. handover-rule errors that bypass silent_ingest).
+            # Gated by ``WHATSAPP_FORWARD_OWNER_MESSAGES`` at the bridge layer;
+            # metadata + text tagging are unconditional when the flag is present
+            # so a future producer can set it without adapter changes.
             if data.get("fromOwner"):
                 metadata["whatsapp_from_owner"] = True
+                if not body.startswith(_OWNER_REPLY_PREFIX):
+                    body = f"{_OWNER_REPLY_PREFIX}{body}"
 
             return MessageEvent(
                 text=body,

tests/gateway/test_whatsapp_from_owner.py

@@ -1,11 +1,12 @@
-"""Tests for WhatsApp owner-message metadata propagation.
+"""Tests for WhatsApp owner-message metadata and source-level text tagging.
 
 The Node bridge sets ``fromOwner: true`` on inbound `fromMe` messages that
 look owner-typed (linked-device send, not echoed from /send) when the
 operator opts into ``WHATSAPP_FORWARD_OWNER_MESSAGES``.  These tests pin
 the adapter's responsibility: lift that flag onto
-``MessageEvent.metadata["whatsapp_from_owner"]`` and otherwise leave it
-absent.  The env-var gate itself lives in the bridge — the adapter just
+``MessageEvent.metadata["whatsapp_from_owner"]``, prefix ``MessageEvent.text``
+with ``[owner reply] ``, and otherwise leave metadata absent and text
+unchanged.  The env-var gate itself lives in the bridge — the adapter just
 trusts the payload.
 """
 
@@ -64,6 +65,36 @@ def test_metadata_flag_set_when_payload_has_from_owner():
 
     assert event is not None
     assert event.metadata.get("whatsapp_from_owner") is True
+    assert event.text.startswith("[owner reply] ")
+    assert event.text == "[owner reply] hi from the linked phone"
+
+
+def test_from_owner_does_not_double_prefix_when_already_tagged():
+    adapter = _make_adapter()
+    payload = _dm_payload(
+        fromOwner=True,
+        body="[owner reply] already tagged",
+    )
+
+    event = asyncio.run(adapter._build_message_event(payload))
+
+    assert event is not None
+    assert event.metadata.get("whatsapp_from_owner") is True
+    assert event.text == "[owner reply] already tagged"
+
+
+def test_from_owner_prefixes_empty_body_for_uniform_media_placeholders():
+    """Owner media with empty caption still gets the marker (bridge may
+    substitute placeholders like ``[image received]`` upstream; empty stays
+    tagged for consistency)."""
+    adapter = _make_adapter()
+    payload = _dm_payload(fromOwner=True, body="")
+
+    event = asyncio.run(adapter._build_message_event(payload))
+
+    assert event is not None
+    assert event.metadata.get("whatsapp_from_owner") is True
+    assert event.text == "[owner reply] "
 
 
 def test_metadata_flag_absent_by_default():