fix: suppress final response after media delivery

AGENTS.md

@@ -17,7 +17,11 @@ plugin.
   final network client rather than replacing the host handler.
 - Plugins must use `MediaPayload` + `deliver_media` for attachments. The kit
   owns Hermes media directives, task-local `origin` resolution, route redaction,
-  and the typed result; consumers must not recreate those contracts.
+  the typed result, and successful-send final-response suppression. Consumers
+  must register `transform_media_delivery_output` as Hermes'
+  `transform_llm_output` hook and `clear_media_delivery_state` as
+  `on_session_end`; they must not recreate those contracts or substitute
+  OpenClaw response shapes.
 - Use `tool_name(namespace, verb, noun)` for new tools and prefer explicit
   verbs such as `read`, `write`, and `patch`. Do not use Hermes agent-loop
   names (`memory`, `todo`, `session_search`, `delegate_task`) as plugin tools.

README.md

@@ -227,6 +227,28 @@ plugin never imports gateway internals or exposes raw group IDs to the model.
 The returned `MediaDeliveryResult` carries success, media type, path, requested
 route, a privacy-safe display route, and a redacted host result.
 
+Direct delivery and final response delivery are separate stages in Hermes. A
+consumer that calls `deliver_media` must register the kit's matching Hermes
+lifecycle hooks so a successful direct send cannot be followed by model-authored
+text or a duplicate `MEDIA:` directive:
+
+```python
+from hermes_plugin_kit import (
+    clear_media_delivery_state,
+    transform_media_delivery_output,
+)
+
+def register(ctx):
+    ctx.register_hook("transform_llm_output", transform_media_delivery_output)
+    ctx.register_hook("on_session_end", clear_media_delivery_state)
+```
+
+Successful delivery arms a one-turn marker for the current Hermes session.
+`transform_media_delivery_output` consumes it and returns Hermes' canonical
+`NO_REPLY` response before the gateway sees the final text. Failed delivery does
+not arm suppression, and `on_session_end` clears any unconsumed marker. These are
+Hermes-agent lifecycle and response contracts; they are not OpenClaw shapes.
+
 For non-media host calls, `invoke_host_tool` remains the lower-level seam.
 `invoke_host_tool` resolves the supported direct host handler and wraps the nested
 operation with Hermes `pre_tool_call` and `post_tool_call` hooks. A blocking hook

hermes_plugin_kit/__init__.py

@@ -52,6 +52,7 @@ def register(ctx):
 import logging
 import re
 import sys
+import threading
 import time
 from dataclasses import dataclass
 from enum import Enum
@@ -66,6 +67,8 @@ def register(ctx):
     "invoke_host_tool",
     "deliver_media",
     "resolve_delivery_target",
+    "transform_media_delivery_output",
+    "clear_media_delivery_state",
     "MediaType",
     "MediaPayload",
     "ResolvedDeliveryTarget",
@@ -100,6 +103,9 @@ def register(ctx):
     "turn_id",
     "api_request_id",
 )
+_MEDIA_DELIVERY_STATE_TTL_SECONDS = 300.0
+_MEDIA_DELIVERY_STATE: dict[str, float] = {}
+_MEDIA_DELIVERY_STATE_LOCK = threading.Lock()
 
 
 @dataclass(frozen=True)
@@ -660,6 +666,91 @@ def redact(value: Any) -> Any:
     return safe if isinstance(safe, dict) else {"result": safe}
 
 
+def _media_delivery_context_ids(context: dict[str, Any]) -> tuple[str, ...]:
+    return tuple(
+        dict.fromkeys(
+            str(context.get(key) or "").strip()
+            for key in ("session_id", "task_id")
+            if str(context.get(key) or "").strip()
+        )
+    )
+
+
+def _prune_media_delivery_state(now: float) -> None:
+    expired = [
+        key
+        for key, created_at in _MEDIA_DELIVERY_STATE.items()
+        if now - created_at > _MEDIA_DELIVERY_STATE_TTL_SECONDS
+    ]
+    for key in expired:
+        _MEDIA_DELIVERY_STATE.pop(key, None)
+
+
+def _mark_media_delivery_success(context: dict[str, Any]) -> None:
+    context_ids = _media_delivery_context_ids(context)
+    if not context_ids:
+        return
+    now = time.monotonic()
+    with _MEDIA_DELIVERY_STATE_LOCK:
+        _prune_media_delivery_state(now)
+        for context_id in context_ids:
+            _MEDIA_DELIVERY_STATE[context_id] = now
+    logging.getLogger("hermes_plugin_kit").info(
+        "event=media_final_response_suppression_armed"
+    )
+
+
+def transform_media_delivery_output(
+    response_text: str = "",
+    session_id: str = "",
+    **_: Any,
+) -> str | None:
+    """Hermes ``transform_llm_output`` hook for an already-delivered attachment.
+
+    A successful :func:`deliver_media` call arms one task-local suppression.
+    The finalizer then replaces any model-authored text or repeated ``MEDIA``
+    tag with Hermes' canonical ``NO_REPLY`` marker. State is consumed once, so
+    later turns in the same session are unaffected.
+    """
+    context_id = str(session_id or "").strip()
+    if not context_id:
+        return None
+    now = time.monotonic()
+    with _MEDIA_DELIVERY_STATE_LOCK:
+        _prune_media_delivery_state(now)
+        armed = _MEDIA_DELIVERY_STATE.pop(context_id, None) is not None
+    if not armed:
+        return None
+    logging.getLogger("hermes_plugin_kit").info(
+        "event=media_final_response_suppressed original_chars=%d",
+        len(str(response_text or "")),
+    )
+    return "NO_REPLY"
+
+
+def clear_media_delivery_state(
+    session_id: str = "",
+    task_id: str = "",
+    **_: Any,
+) -> None:
+    """Hermes ``on_session_end`` cleanup for an unconsumed suppression marker."""
+    context_ids = tuple(
+        dict.fromkeys(
+            value
+            for value in (
+                str(session_id or "").strip(),
+                str(task_id or "").strip(),
+            )
+            if value
+        )
+    )
+    if not context_ids:
+        return
+    with _MEDIA_DELIVERY_STATE_LOCK:
+        for context_id in context_ids:
+            _MEDIA_DELIVERY_STATE.pop(context_id, None)
+
+
 def deliver_media(
     media: MediaPayload,
     *,
@@ -705,6 +796,8 @@ def deliver_media(
         media.media_type.value,
         success,
     )
+    if success:
+        _mark_media_delivery_success(context)
     return MediaDeliveryResult(
         success=success,
         requested_target=(

pyproject.toml

@@ -8,7 +8,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hermes-plugin-kit"
-version = "0.2.0"
+version = "0.2.1"
 description = "Convention-correct lifecycle registration for hermes-agent plugins."
 readme = "README.md"
 requires-python = ">=3.11"

tests/test_kit.py

@@ -320,6 +320,10 @@ def test_rejects_unknown_host_tool(self) -> None:
 
 
 class MediaDeliveryContractTests(unittest.TestCase):
+    def tearDown(self) -> None:
+        hpk.clear_media_delivery_state(session_id="session-1")
+        hpk.clear_media_delivery_state(session_id="session-2")
+
     def test_voice_payload_has_typed_hermes_directive(self) -> None:
         payload = hpk.MediaPayload("/opt/data/voice-staging/memo.ogg", hpk.MediaType.VOICE)
         self.assertEqual(
@@ -428,6 +432,107 @@ def test_deliver_media_invokes_typed_host_contract(self) -> None:
         self.assertEqual(result.host_result["chat_id"], "…2527")
         self.assertEqual(result.as_dict()["media_type"], "voice")
 
+    def test_successful_delivery_suppresses_the_same_turn_final_response_once(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "memo.ogg"
+            path.write_bytes(b"OggS" + b"\x00" * 16)
+            with patch.object(
+                hpk,
+                "invoke_host_tool",
+                return_value=json.dumps({"success": True, "message_id": "9"}),
+            ):
+                hpk.deliver_media(
+                    hpk.MediaPayload(path, hpk.MediaType.VOICE),
+                    target="telegram:8670382527",
+                    session_id="session-1",
+                )
+
+        self.assertEqual(
+            hpk.transform_media_delivery_output(
+                response_text=f"[[audio_as_voice]]\nMEDIA:{path}",
+                session_id="session-1",
+                platform="telegram",
+            ),
+            "NO_REPLY",
+        )
+        self.assertIsNone(
+            hpk.transform_media_delivery_output(
+                response_text="unrelated next turn",
+                session_id="session-1",
+                platform="telegram",
+            )
+        )
+
+    def test_successful_delivery_does_not_suppress_another_session(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "memo.ogg"
+            path.write_bytes(b"OggS" + b"\x00" * 16)
+            with patch.object(
+                hpk,
+                "invoke_host_tool",
+                return_value=json.dumps({"success": True}),
+            ):
+                hpk.deliver_media(
+                    hpk.MediaPayload(path, hpk.MediaType.VOICE),
+                    target="telegram:-5372910000",
+                    session_id="session-1",
+                )
+
+        self.assertIsNone(
+            hpk.transform_media_delivery_output(
+                response_text="keep this",
+                session_id="session-2",
+                platform="telegram",
+            )
+        )
+
+    def test_failed_delivery_does_not_suppress_final_response(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "memo.ogg"
+            path.write_bytes(b"OggS" + b"\x00" * 16)
+            with patch.object(
+                hpk,
+                "invoke_host_tool",
+                return_value=json.dumps({"success": False, "error": "offline"}),
+            ):
+                hpk.deliver_media(
+                    hpk.MediaPayload(path, hpk.MediaType.VOICE),
+                    target="telegram:8670382527",
+                    session_id="session-1",
+                )
+
+        self.assertIsNone(
+            hpk.transform_media_delivery_output(
+                response_text="delivery failed",
+                session_id="session-1",
+                platform="telegram",
+            )
+        )
+
+    def test_session_end_clears_unconsumed_delivery_state(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "memo.ogg"
+            path.write_bytes(b"OggS" + b"\x00" * 16)
+            with patch.object(
+                hpk,
+                "invoke_host_tool",
+                return_value=json.dumps({"success": True}),
+            ):
+                hpk.deliver_media(
+                    hpk.MediaPayload(path, hpk.MediaType.VOICE),
+                    target="telegram:8670382527",
+                    session_id="session-1",
+                )
+
+        hpk.clear_media_delivery_state(session_id="session-1")
+        self.assertIsNone(
+            hpk.transform_media_delivery_output(
+                response_text="next turn",
+                session_id="session-1",
+                platform="telegram",
+            )
+        )
+
     def test_delivery_result_redacts_raw_route_from_host_errors(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             path = Path(tmp) / "memo.ogg"