fix(webhook): make intake fan-out-safe and structurally valid (Webhook Revolution)

gateway/platforms/webhook.py

@@ -63,6 +63,10 @@
     DEFAULT_SCRIPT_TIMEOUT_SECONDS,
     WebhookRouteProcessor,
 )
+from gateway.platforms.webhook_profile_admission import (
+    WebhookProfileAdmissionMixin,
+    _PROFILE_REJECTED,
+)
 from gateway.response_filters import is_autonomous_silence_response
 
 logger = logging.getLogger(__name__)
@@ -96,11 +100,6 @@ def _is_webhook_silence_response(content: Any) -> bool:
     """
     return is_autonomous_silence_response(content)
 
-# Sentinel returned by _resolve_request_profile when a /p/<profile>/ prefix
-# names a profile this gateway does not serve (→ 404). Distinct from None
-# (no prefix / multiplexing off → handle as the default profile).
-_PROFILE_REJECTED = object()
-
 _BUILTIN_DELIVER_PLATFORMS = {
     "telegram", "discord", "slack", "signal", "sms", "whatsapp",
     "matrix", "mattermost", "homeassistant", "email", "dingtalk",
@@ -174,7 +173,7 @@ def check_webhook_requirements() -> bool:
     return AIOHTTP_AVAILABLE
 
 
-class WebhookAdapter(BasePlatformAdapter):
+class WebhookAdapter(WebhookProfileAdmissionMixin, BasePlatformAdapter):
     """Generic webhook receiver that triggers agent runs from HTTP POSTs."""
 
     # No human is present to answer a "session restored — what next?" prompt:
@@ -219,7 +218,10 @@ def __init__(self, config: PlatformConfig):
 
         # Idempotency: TTL cache of recently processed delivery IDs.
         # Prevents duplicate agent runs when webhook providers retry.
-        self._seen_deliveries: Dict[str, float] = {}
+        # Keyed by (profile, route, delivery_id); bound to a body hash so a
+        # conflicting replay (same key, different body) can be reported as 409.
+        self._seen_deliveries: Dict[tuple, float] = {}
+        self._seen_delivery_bodies: Dict[tuple, str] = {}
         self._idempotency_ttl: int = 3600  # 1 hour
         self._seen_deliveries_next_prune_at: float = 0.0
 
@@ -431,6 +433,7 @@ def _prune_seen_deliveries(self, now: float) -> None:
         stale = [k for k, t in self._seen_deliveries.items() if t < cutoff]
         for k in stale:
             self._seen_deliveries.pop(k, None)
+            self._seen_delivery_bodies.pop(k, None)
         self._seen_deliveries_next_prune_at = now + min(60.0, max(1.0, self._idempotency_ttl / 10))
 
     def _record_rate_limit_hit(self, route_name: str, now: float) -> bool:
@@ -448,14 +451,60 @@ def _record_rate_limit_hit(self, route_name: str, now: float) -> bool:
         window.append(now)
         return True
 
-    def _record_delivery_id(self, delivery_id: str, now: float) -> bool:
-        """Return True when this delivery should be processed."""
-        seen_at = self._seen_deliveries.get(delivery_id)
-        if seen_at is not None and now - seen_at < self._idempotency_ttl:
+    def _profile_scope_key(self) -> str:
+        """Return the current profile scope (or 'default') for idempotency keys."""
+        runner = getattr(self, "gateway_runner", None)
+        active = getattr(runner, "_active_profile_name", None)
+        if callable(active):
+            try:
+                val = active()
+            except Exception:
+                val = None
+            if val:
+                return str(val)
+        return getattr(self, "_webhook_profile", "default")
+
+    def _active_route_key(self) -> str:
+        """Return the active route name (or '') for idempotency keys."""
+        return getattr(self, "_webhook_active_route", "")
+
+    def _record_delivery_id(
+        self,
+        delivery_id: str,
+        now: float,
+        body_hash: str = "",
+        *,
+        profile: str | None = None,
+        route: str | None = None,
+    ) -> bool:
+        """Return True when this delivery should be processed.
+
+        Idempotency is keyed by ``(profile, route, provider, delivery_id)``
+        and bound to a body hash. A retry of the SAME delivery on the same
+        route is suppressed; the same provider delivery intentionally sent to
+        DIFFERENT routes executes each route once (#7448). Conflicting reuse
+        (same key, different body) is reported via the return sentinel so the
+        handler can emit 409.
+        """
+        key = (
+            profile or self._profile_scope_key(),
+            route or self._active_route_key(),
+            delivery_id,
+        )
+        entry = self._seen_deliveries.get(key)
+        if entry is not None and now - entry < self._idempotency_ttl:
+            # Same key replayed. If a body hash was bound and differs, the
+            # caller should treat this as a conflict (409), not a duplicate.
+            if entry_body := self._seen_delivery_bodies.get(key):
+                if body_hash and entry_body != body_hash:
+                    return "conflict"  # type: ignore[return-value]
             return False
-        if seen_at is not None:
-            self._seen_deliveries.pop(delivery_id, None)
-        self._seen_deliveries[delivery_id] = now
+        if entry is not None:
+            self._seen_deliveries.pop(key, None)
+            self._seen_delivery_bodies.pop(key, None)
+        self._seen_deliveries[key] = now
+        if body_hash:
+            self._seen_delivery_bodies[key] = body_hash
         if len(self._seen_deliveries) > max(self._rate_limit * 2, 128):
             self._prune_seen_deliveries(now)
         return True
@@ -530,65 +579,6 @@ def _reload_dynamic_routes(self) -> None:
         except Exception as e:
             logger.error("[webhook] Failed to reload dynamic routes: %s", e)
 
-    def _resolve_request_profile(self, request: "web.Request"):
-        """Resolve + validate the /p/<profile>/ URL prefix on a webhook request.
-
-        Returns:
-          - ``None`` when no profile prefix is present, or multiplexing is off
-            (the prefix is ignored, request handled as the default profile).
-          - the profile name (str) when present, multiplexing is on, and the
-            profile is one this gateway serves.
-          - ``_PROFILE_REJECTED`` when a prefix is present but the profile is
-            unknown/unconfigured (handler returns 404).
-        """
-        profile = (request.match_info.get("profile") or "").strip()
-        if not profile:
-            return None
-        runner = self.gateway_runner
-        cfg = getattr(runner, "config", None)
-        if not getattr(cfg, "multiplex_profiles", False):
-            # Prefix supplied but multiplexing is off — ignore it, behave as
-            # the single-profile gateway (don't 404 a would-be valid route).
-            return None
-        try:
-            from hermes_cli.profiles import profiles_to_serve
-            served = {
-                name
-                for name, _ in profiles_to_serve(
-                    multiplex=True,
-                    profile_allowlist=getattr(
-                        cfg, "multiplex_profile_allowlist", None
-                    ),
-                )
-            }
-        except Exception:
-            return _PROFILE_REJECTED
-        if profile not in served:
-            return _PROFILE_REJECTED
-        return profile
-
-    @staticmethod
-    def _route_allows_profile(
-        route_config: dict,
-        request_profile: Optional[str],
-    ) -> bool:
-        """Return whether a route is bound to the URL-selected profile.
-
-        Omitting ``profile`` keeps a route on the default profile. An explicit
-        null, blank, or non-string value is malformed and fails closed.
-        """
-        if "profile" not in route_config:
-            configured_profile = "default"
-        else:
-            configured_profile = route_config.get("profile")
-        if not isinstance(configured_profile, str):
-            return False
-        configured_profile = configured_profile.strip()
-        if not configured_profile:
-            return False
-        effective_profile = request_profile or "default"
-        return configured_profile == effective_profile
-
     async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         """POST /webhooks/{route_name} — receive and process a webhook event."""
         # Hot-reload dynamic subscriptions on each request (mtime-gated, cheap)
@@ -685,7 +675,9 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         now = time.time()
         if not self._record_rate_limit_hit(route_name, now):
             return web.json_response(
-                {"error": "Rate limit exceeded"}, status=429
+                {"error": "Rate limit exceeded"},
+                status=429,
+                headers={"Retry-After": str(_RATE_WINDOW_SECONDS)},
             )
 
         # Parse payload
@@ -808,9 +800,29 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         )
 
         # ── Idempotency ─────────────────────────────────────────
-        # Skip duplicate deliveries (webhook retries).
+        # Skip duplicate deliveries (webhook retries). Keyed by
+        # (profile, route, delivery_id) and bound to a body hash so the same
+        # provider delivery sent to different routes executes each route once
+        # (#7448), while a conflicting replay on the same route returns 409.
         now = time.time()
-        if not self._record_delivery_id(delivery_id, now):
+        body_hash = hashlib.sha256(raw_body).hexdigest()
+        idem_result = self._record_delivery_id(
+            delivery_id,
+            now,
+            body_hash,
+            profile=profile or "default",
+            route=route_name,
+        )
+        if idem_result == "conflict":
+            return web.json_response(
+                {
+                    "status": "conflict",
+                    "delivery_id": delivery_id,
+                    "error": "Idempotency key reused with a different body",
+                },
+                status=409,
+            )
+        if not idem_result:
             logger.info(
                 "[webhook] Skipping duplicate delivery %s", delivery_id
             )
@@ -1214,9 +1226,12 @@ def _render_prompt(
         Supports dot-notation access into nested dicts:
         ``{pull_request.title}`` → ``payload["pull_request"]["title"]``
 
-        Special token ``{__raw__}`` dumps the entire payload as indented
-        JSON (truncated to 4000 chars).  Useful for monitoring alerts or
-        any webhook where the agent needs to see the full payload.
+        Special token ``{__raw__}`` dumps the entire payload as a valid JSON
+        envelope ``{"payload": <value>, "truncated": <bool>,
+        "original_bytes": <N>}``.  When the payload exceeds the bounded cap,
+        the envelope is still structurally valid JSON so a downstream agent or
+        tool can parse it without hitting a truncated/raw character slice
+        (#55829).
         """
         if not template:
             truncated = json.dumps(payload, indent=2)[:4000]
@@ -1227,9 +1242,9 @@ def _render_prompt(
 
         def _resolve(match: re.Match) -> str:
             key = match.group(1)
-            # Special token: dump the entire payload as JSON
+            # Special token: dump the entire payload as a valid JSON envelope
             if key == "__raw__":
-                return json.dumps(payload, indent=2)[:4000]
+                return self._render_raw_payload(payload)
             if key == "event_type":
                 return event_type
             value: Any = payload
@@ -1244,6 +1259,26 @@ def _resolve(match: re.Match) -> str:
 
         return re.sub(r"\{([a-zA-Z0-9_.]+)\}", _resolve, template)
 
+    def _render_raw_payload(self, payload: dict, cap: int = 4000) -> str:
+        """Render ``{__raw__}`` as a structurally valid JSON envelope.
+
+        ``original_bytes`` is the serialized payload length; when the payload
+        exceeds ``cap`` the ``payload`` field holds the bounded truncated
+        value and ``truncated`` is True, so the output always parses.
+        """
+        serialized = json.dumps(payload, indent=2, ensure_ascii=False)
+        original_bytes = len(serialized)
+        truncated = original_bytes > cap
+        bounded = serialized[:cap]
+        return json.dumps(
+            {
+                "payload": bounded,
+                "truncated": truncated,
+                "original_bytes": original_bytes,
+            },
+            ensure_ascii=False,
+        )
+
     def _render_delivery_extra(
         self, extra: dict, payload: dict
     ) -> dict:

gateway/platforms/webhook_profile_admission.py

@@ -0,0 +1,69 @@
+"""Profile admission policy for the generic webhook adapter."""
+
+from typing import Optional
+
+try:
+    from aiohttp import web
+except ImportError:
+    web = None  # type: ignore[assignment]
+
+
+# Sentinel returned by _resolve_request_profile when a /p/<profile>/ prefix
+# names a profile this gateway does not serve (→ 404). Distinct from None
+# (no prefix / multiplexing off → handle as the default profile).
+_PROFILE_REJECTED = object()
+
+
+class WebhookProfileAdmissionMixin:
+    """Resolve and authorize profile-bound webhook requests."""
+
+    def _resolve_request_profile(self, request: "web.Request"):
+        """Resolve + validate the /p/<profile>/ URL prefix on a webhook request.
+
+        Returns:
+          - ``None`` when no profile prefix is present, or multiplexing is off
+            (the prefix is ignored, request handled as the default profile).
+          - the profile name (str) when present, multiplexing is on, and the
+            profile is one this gateway serves.
+          - ``_PROFILE_REJECTED`` when a prefix is present but the profile is
+            unknown/unconfigured (handler returns 404).
+        """
+        profile = (request.match_info.get("profile") or "").strip()
+        if not profile:
+            return None
+        runner = self.gateway_runner
+        cfg = getattr(runner, "config", None)
+        if not getattr(cfg, "multiplex_profiles", False):
+            # Prefix supplied but multiplexing is off — ignore it, behave as
+            # the single-profile gateway (don't 404 a would-be valid route).
+            return None
+        try:
+            from hermes_cli.profiles import profiles_to_serve
+            served = {name for name, _ in profiles_to_serve(multiplex=True)}
+        except Exception:
+            return _PROFILE_REJECTED
+        if profile not in served:
+            return _PROFILE_REJECTED
+        return profile
+
+    @staticmethod
+    def _route_allows_profile(
+        route_config: dict,
+        request_profile: Optional[str],
+    ) -> bool:
+        """Return whether a route is bound to the URL-selected profile.
+
+        Omitting ``profile`` keeps a route on the default profile. An explicit
+        null, blank, or non-string value is malformed and fails closed.
+        """
+        if "profile" not in route_config:
+            configured_profile = "default"
+        else:
+            configured_profile = route_config.get("profile")
+        if not isinstance(configured_profile, str):
+            return False
+        configured_profile = configured_profile.strip()
+        if not configured_profile:
+            return False
+        effective_profile = request_profile or "default"
+        return configured_profile == effective_profile