From ba424b35956db3201b7866a100673ede94eb4ddb Mon Sep 17 00:00:00 2001 From: Tharick Jairam Date: Fri, 7 Aug 2026 14:19:36 -0400 Subject: [PATCH 1/3] Add Facebook comment and reaction webhook handling. Introduce comment/reaction adapters and filters, extend the Graph API helpers and webhook routing so page comments and reactions can be ingested alongside existing Messenger message flows. Co-authored-by: Cursor --- jvagent/action/facebook_action/endpoints.py | 166 ++++++++++++- .../action/facebook_action/facebook_action.py | 126 +++++++++- .../action/facebook_action/facebook_api.py | 227 ++++++++++++++++++ .../facebook_comment_adapter.py | 168 +++++++++++++ .../facebook_comment_filter.py | 54 +++++ .../facebook_reaction_adapter.py | 38 +++ .../messenger_webhook_helpers.py | 198 +++++++++++++++ 7 files changed, 970 insertions(+), 7 deletions(-) create mode 100644 jvagent/action/facebook_action/facebook_comment_adapter.py create mode 100644 jvagent/action/facebook_action/facebook_comment_filter.py create mode 100644 jvagent/action/facebook_action/facebook_reaction_adapter.py diff --git a/jvagent/action/facebook_action/endpoints.py b/jvagent/action/facebook_action/endpoints.py index 2af99e80..d3288943 100644 --- a/jvagent/action/facebook_action/endpoints.py +++ b/jvagent/action/facebook_action/endpoints.py @@ -20,7 +20,12 @@ from .facebook_api import FacebookAPI from .messenger_message_coalescer import MessengerMessageCoalescer from .messenger_webhook_helpers import ( + FEED_COMMENT_UTTERANCE_MAX, + FEED_REACTION_UTTERANCE_MAX, + _synthesize_reaction_utterance, prime_messenger_sender_actions, + process_feed_comment_interaction_async, + process_feed_reaction_interaction_async, process_messenger_interaction_async, resolve_messenger_inbound_event, verify_meta_messenger_signature, @@ -664,7 +669,10 @@ async def messenger_interact_webhook_events(request: Request, agent_id: str) -> raise HTTPException(status_code=400, detail="Invalid JSON body") events = FacebookAPI.iter_messenger_user_text_events(payload) - if not events: + feed_comment_events = FacebookAPI.iter_feed_comment_events(payload) + feed_reaction_events = FacebookAPI.iter_feed_reaction_events(payload) + + if not events and not feed_comment_events and not feed_reaction_events: return {"status": "ignored", "response": None} access_control_action = await agent.get_access_control_action() @@ -748,6 +756,126 @@ async def handle_merged_messenger_event(merged_event: Dict[str, Any]) -> None: handle_merged_messenger_event, ) + # Process feed/comment events (Facebook Page comments) + for comment_event in feed_comment_events: + sender = str(comment_event.get("sender_id") or "").strip() + if not sender: + continue + page_id = str(comment_event.get("page_id") or "").strip() + # Defense-in-depth: skip if sender is the Page itself (should already + # be filtered by iter_feed_comment_events, but guard here too). + if sender and page_id and sender == page_id: + logger.debug( + "Skipping feed comment from Page %s (self-reply loop prevention)", + sender, + ) + continue + comment_text = str(comment_event.get("message") or "").strip() + if not comment_text: + continue + if len(comment_text) > FEED_COMMENT_UTTERANCE_MAX: + continue + + comment_id = str(comment_event.get("comment_id") or "").strip() + post_id = str(comment_event.get("post_id") or "").strip() + sender_name = str(comment_event.get("sender_name") or "").strip() or None + + feed_data_dict: Dict[str, Any] = { + "feed_payload": comment_event, + "facebook_comment": True, + "post_id": post_id, + "comment_id": comment_id, + "page_id": page_id, + } + + has_access = True + if access_control_action: + has_access = await access_control_action.has_action_access( + user_id=sender, + action_label="FacebookAction", + channel="facebook_comment", + ) + if not has_access: + log_access_denied( + agent_id=agent_id, + user_id=sender, + channel="facebook_comment", + action_label="FacebookAction", + stage="feed_comment", + ) + continue + + await process_feed_comment_interaction_async( + comment_text, + sender, + agent_id, + agent, + feed_data_dict, + sender_name=sender_name, + ) + + # Process feed/reaction events (Facebook Page reactions) + for reaction_event in feed_reaction_events: + sender = str(reaction_event.get("sender_id") or "").strip() + if not sender: + continue + page_id = str(reaction_event.get("page_id") or "").strip() + # Defense-in-depth: skip if sender is the Page itself. + if sender and page_id and sender == page_id: + logger.debug( + "Skipping feed reaction from Page %s (self-loop prevention)", + sender, + ) + continue + reaction_type = str(reaction_event.get("reaction_type") or "").strip() + if not reaction_type: + continue + + post_id = str(reaction_event.get("post_id") or "").strip() + comment_id = str(reaction_event.get("comment_id") or "").strip() + sender_name = str(reaction_event.get("sender_name") or "").strip() or None + + utterance = _synthesize_reaction_utterance( + reaction_type, sender_name or "", comment_id, post_id + ) + if len(utterance) > FEED_REACTION_UTTERANCE_MAX: + utterance = utterance[: FEED_REACTION_UTTERANCE_MAX - 3] + "..." + + reaction_data_dict: Dict[str, Any] = { + "feed_payload": reaction_event, + "facebook_reaction": True, + "post_id": post_id, + "comment_id": comment_id, + "reaction_type": reaction_type, + "page_id": page_id, + } + + has_access = True + if access_control_action: + has_access = await access_control_action.has_action_access( + user_id=sender, + action_label="FacebookAction", + channel="facebook_reaction", + ) + if not has_access: + log_access_denied( + agent_id=agent_id, + user_id=sender, + channel="facebook_reaction", + action_label="FacebookAction", + stage="feed_reaction", + ) + continue + + await process_feed_reaction_interaction_async( + utterance, + sender, + agent_id, + agent, + reaction_data_dict, + sender_name=sender_name, + ) + return {"status": "received"} @@ -805,3 +933,39 @@ async def facebook_register_webhook( ) _raise_if_graph_error(action_id, result) return {"success": True, "result": result} + + +@endpoint( + "/actions/{action_id}/facebook/page/subscribe-app", + methods=["POST"], + auth=True, + roles=["admin"], + tags=["Facebook Action"], + summary="Install the app on the Page to receive feed/comment webhook events", +) +async def facebook_subscribe_app_to_page( + action_id: str, +) -> Dict[str, Any]: + """Subscribe the app to the Facebook Page so it receives Page-level events. + + This is required in addition to the app-level webhook subscription + (``/actions/{action_id}/facebook/webhook/register``). Meta requires both: + + 1. App-level subscription (``/{app_id}/subscriptions``) — registers + which fields (feed, messages, etc.) the webhook should receive. + 2. Per-page installation (``/{page_id}/subscribed_apps``) — tells Meta + to deliver Page events for this specific Page. + + Without step 2, ``feed`` events (comments, reactions) are not delivered + even if ``feed`` is in the webhook subscription fields. + + Requires ``pages_manage_metadata`` permission on the Page access token. + """ + action = await _require_facebook_action(action_id) + result = await action.subscribe_app_to_page() + if result.get("status") == "error": + raise ValidationError( + message=f"Failed to subscribe app to Page: {result.get('error', 'unknown')}", + details={"action_id": action_id, "result": result}, + ) + return result diff --git a/jvagent/action/facebook_action/facebook_action.py b/jvagent/action/facebook_action/facebook_action.py index b2db9d43..92c10198 100644 --- a/jvagent/action/facebook_action/facebook_action.py +++ b/jvagent/action/facebook_action/facebook_action.py @@ -580,6 +580,56 @@ def _register() -> Dict[str, Any]: ) return {"status": "error", "error": str(e)} + async def subscribe_app_to_page(self) -> Dict[str, Any]: + """Install the app on the Page so it receives Page-level webhook events. + + Meta requires two steps for Page events (feed, comments, reactions): + 1. App-level subscription via ``register_messenger_webhook_subscription`` + (``POST /{app_id}/subscriptions``). + 2. Per-page installation via this method (``POST /{page_id}/subscribed_apps``). + + Without this second step, ``feed`` webhook events (comments on Page posts) + are not delivered even if ``feed`` is in the subscribed fields. + + Requires ``pages_manage_metadata`` permission on the Page access token. + """ + self._apply_env_defaults() + if not self.is_configured(): + return { + "status": "skipped", + "reason": "Facebook action is not configured", + "issues": self._config_issues(), + } + page_token = self._page_access_token() + if not page_token: + await self._maybe_resolve_page_access_token() + page_token = self._page_access_token() + if not page_token: + return { + "status": "skipped", + "reason": "No page access token available (set FACEBOOK_PAGE_ACCESS_TOKEN or ensure user token has pages_show_list)", + } + try: + + def _subscribe() -> Dict[str, Any]: + return self.api().subscribe_app_to_page() + + result = await asyncio.to_thread(_subscribe) + if isinstance(result, dict) and result.get("error"): + logger.warning( + "Meta app-to-page subscription Graph error: %s", + result.get("error"), + ) + else: + logger.info( + "App subscribed to Page (page_id=%s) — feed/comment events enabled", + self.page_id, + ) + return {"status": "ok", "page_id": self.page_id, "result": result} + except Exception as e: + logger.error("subscribe_app_to_page failed: %s", e, exc_info=True) + return {"status": "error", "error": str(e)} + async def on_register(self) -> None: self._apply_env_defaults() if self._base_graph_config_issues(): @@ -587,6 +637,9 @@ async def on_register(self) -> None: return await self._maybe_resolve_page_access_token() await self._ensure_messenger_webhook_url() + sub = await self.subscribe_app_to_page() + if sub.get("status") not in ("ok", "skipped"): + logger.warning("FacebookAction on_register: subscribe_app_to_page: %s", sub) logger.debug("Facebook action registered") async def on_reload(self) -> None: @@ -618,6 +671,11 @@ async def on_reload(self) -> None: "FacebookAction on_reload: register_messenger_webhook_subscription: %s", reg, ) + sub = await self.subscribe_app_to_page() + if sub.get("status") not in ("ok", "skipped"): + logger.warning( + "FacebookAction on_reload: subscribe_app_to_page: %s", sub + ) async def on_startup(self) -> None: """Register Messenger response-bus filter and adapter when configured.""" @@ -659,6 +717,21 @@ async def on_startup(self) -> None: if not await MessengerAdapter(action=self).initialize(agent=agent): logger.error("MessengerAdapter initialization failed") + from .facebook_comment_adapter import FacebookCommentAdapter + from .facebook_comment_filter import FacebookCommentFilter + from .facebook_reaction_adapter import FacebookReactionAdapter + + if not await FacebookCommentFilter( + channels=["facebook_comment"], priority=100 + ).initialize(agent=agent): + logger.warning("FacebookCommentFilter initialization failed") + + if not await FacebookCommentAdapter(action=self).initialize(agent=agent): + logger.error("FacebookCommentAdapter initialization failed") + + if not await FacebookReactionAdapter(action=self).initialize(agent=agent): + logger.error("FacebookReactionAdapter initialization failed") + skip_reg = ( os.environ.get("FACEBOOK_SKIP_STARTUP_WEBHOOK_REGISTRATION", "").lower() == "true" @@ -718,6 +791,17 @@ async def _run_startup_messenger_webhook_register(self) -> None: "Facebook Messenger webhook registration: %s", reg, ) + + sub = await self.subscribe_app_to_page() + if sub.get("status") == "ok": + logger.info("Facebook app subscribed to Page for feed events") + elif sub.get("status") == "skipped": + logger.info( + "Facebook app-to-page subscription skipped: %s", + sub.get("reason"), + ) + else: + logger.warning("Facebook app-to-page subscription: %s", sub) except Exception as e: logger.warning( "Facebook Messenger startup webhook register failed: %s", @@ -830,7 +914,7 @@ async def _fallback_deferred() -> None: _task.add_done_callback(_BACKGROUND_TASKS.discard) async def ensure_adapter_registered(self) -> bool: - """Ensure Messenger ChannelAdapter is registered (e.g. Lambda cold start).""" + """Ensure Messenger, FacebookComment and FacebookReaction ChannelAdapters are registered.""" if not self.is_configured(): return False try: @@ -840,12 +924,42 @@ async def ensure_adapter_registered(self) -> bool: response_bus = await agent.get_response_bus() if not response_bus: return False - existing = response_bus._channel_adapters.get("messenger") - if existing and getattr(existing, "_initialized", False): - return True - from .messenger_adapter import MessengerAdapter - return await MessengerAdapter(action=self).initialize(agent=agent) + messenger_ok = True + existing_messenger = response_bus._channel_adapters.get("messenger") + if not ( + existing_messenger + and getattr(existing_messenger, "_initialized", False) + ): + from .messenger_adapter import MessengerAdapter + + messenger_ok = await MessengerAdapter(action=self).initialize( + agent=agent + ) + + comment_ok = True + existing_comment = response_bus._channel_adapters.get("facebook_comment") + if not ( + existing_comment and getattr(existing_comment, "_initialized", False) + ): + from .facebook_comment_adapter import FacebookCommentAdapter + + comment_ok = await FacebookCommentAdapter(action=self).initialize( + agent=agent + ) + + reaction_ok = True + existing_reaction = response_bus._channel_adapters.get("facebook_reaction") + if not ( + existing_reaction and getattr(existing_reaction, "_initialized", False) + ): + from .facebook_reaction_adapter import FacebookReactionAdapter + + reaction_ok = await FacebookReactionAdapter(action=self).initialize( + agent=agent + ) + + return bool(messenger_ok and comment_ok and reaction_ok) except Exception as e: logger.error( "FacebookAction: ensure_adapter_registered failed: %s", diff --git a/jvagent/action/facebook_action/facebook_api.py b/jvagent/action/facebook_action/facebook_api.py index 4824c06e..052285be 100644 --- a/jvagent/action/facebook_action/facebook_api.py +++ b/jvagent/action/facebook_action/facebook_api.py @@ -156,6 +156,47 @@ def register_session(self, webhook_url: str) -> Dict: "POST", endpoint, params=params, data=body, json_body=None ) + def subscribe_app_to_page(self) -> Dict: + """Install the app on the Page so it receives Page-level webhook events. + + Meta requires **two** steps for Page events (feed, comments, reactions): + 1. App-level subscription via ``register_session`` (``/{app_id}/subscriptions``). + 2. Per-page installation via this method (``/{page_id}/subscribed_apps``). + + Without this second step, ``feed`` webhook events (comments on Page posts) + are not delivered, even if ``feed`` is in the subscribed fields. + + Uses the **page access token** (requires ``pages_manage_metadata`` permission). + + Returns: + Graph API response dict. On success Meta returns ``{"success": true}``. + """ + endpoint = f"{self.page_id}/subscribed_apps" + fields_val = ( + self.fields + or "feed,messages,messaging_postbacks,message_deliveries,standby,mention" + ) + if isinstance(fields_val, list): + fields_val = ",".join(str(x) for x in fields_val) + params = { + "access_token": self._token_for_page(), + "subscribed_fields": str(fields_val), + } + return self.send_rest_request("POST", endpoint, params=params) + + def list_subscribed_apps(self) -> Union[List, Dict]: + """List apps installed on the Page that receive webhook events. + + Uses the page access token (requires ``pages_manage_metadata`` or + ``pages_read_engagement`` permission). + + Returns: + List of app dicts or error dict. + """ + endpoint = f"{self.page_id}/subscribed_apps" + params = {"access_token": self._token_for_page()} + return self.send_rest_request("GET", endpoint, params=params) + @staticmethod def _messenger_coordinate_value(coords: Any, *keys: str) -> Optional[float]: if not isinstance(coords, dict): @@ -410,6 +451,192 @@ def parse_inbound_message(request: Dict) -> Dict: ) return {"ok": False, "error": str(e)} + @staticmethod + def iter_feed_comment_events( + request: Dict[str, Any], + ) -> List[Dict[str, Any]]: + """Parse Meta Page webhook JSON and return inbound **feed comment** events. + + This is the feed/comment counterpart to + :meth:`iter_messenger_user_text_events`. It processes entries with + ``"changes"`` (feed events) and extracts comment events where + ``item == "comment"``. + + Skips: ``messaging`` entries (DMs), reactions, status changes, and + non-comment feed items. + + Each returned dict includes: + ``sender_id``, ``sender_name``, ``page_id``, ``comment_id``, + ``post_id``, ``message`` (comment text), ``message_type`` (always + ``"comment"``), ``parent_id`` (parent post or comment ID), + ``data`` (raw request payload for downstream channel_gate parsing), + ``timestamp``. + """ + out: List[Dict[str, Any]] = [] + if not isinstance(request, dict): + return out + if request.get("object") != "page": + return out + entries = request.get("entry") + if not isinstance(entries, list): + return out + + for entry in entries: + if not isinstance(entry, dict): + continue + page_id = str(entry.get("id", "")) + changes = entry.get("changes") + if not isinstance(changes, list): + continue + for change in changes: + if not isinstance(change, dict): + continue + value = change.get("value") + if not isinstance(value, dict): + continue + item = str(value.get("item", "")).strip().lower() + if item != "comment": + continue + verb = str(value.get("verb", "")).strip().lower() + if verb not in ("add",): + continue + from_obj = value.get("from") or {} + sender_id = ( + str(from_obj.get("id", "")) if isinstance(from_obj, dict) else "" + ) + sender_name = ( + str(from_obj.get("name", "")) if isinstance(from_obj, dict) else "" + ) + # Skip Page's own comments to prevent self-reply loops. + # When the Page replies via reply_to_comment(), Meta sends a + # webhook event with from.id == page_id — processing it would + # trigger another agent turn and an infinite reply loop. + if sender_id and sender_id == page_id: + FacebookAPI.logger.debug( + "iter_feed_comment_events: skipping Page's own comment " + "sender_id=%s == page_id=%s", + sender_id, + page_id, + ) + continue + message = str(value.get("message", "") or "").strip() + comment_id = str( + value.get("comment_id", "") or value.get("id", "") or "" + ).strip() + post_id = str(value.get("post_id", "") or "").strip() + parent_id = str(value.get("parent_id", "") or post_id or "").strip() + ts = value.get("created_time") or value.get("timestamp") + try: + timestamp = int(ts) if ts is not None else 0 + except (TypeError, ValueError): + timestamp = 0 + out.append( + { + "sender_id": sender_id, + "sender_name": sender_name, + "page_id": page_id, + "comment_id": comment_id, + "post_id": post_id, + "message_type": "comment", + "message": message, + "parent_id": parent_id, + "timestamp": timestamp, + "data": request, + "verb": verb, + } + ) + return out + + @staticmethod + def iter_feed_reaction_events( + request: Dict[str, Any], + ) -> List[Dict[str, Any]]: + """Parse Meta Page webhook JSON and return inbound **feed reaction** events. + + Processes entries with ``"changes"`` (feed events) and extracts + reaction events where ``item == "reaction"`` and ``verb == "add"``. + + Skips: ``messaging`` entries (DMs), comments, status changes, remove + reactions, and the Page's own reactions (self-loop prevention). + + Each returned dict includes: + ``sender_id``, ``sender_name``, ``page_id``, ``post_id``, + ``comment_id`` (empty string if reaction is on a post, not a comment), + ``reaction_type`` (like, love, wow, haha, sorry, anger), + ``message_type`` (always ``"reaction"``), ``data`` (raw request payload), + ``timestamp``. + """ + out: List[Dict[str, Any]] = [] + if not isinstance(request, dict): + return out + if request.get("object") != "page": + return out + entries = request.get("entry") + if not isinstance(entries, list): + return out + + for entry in entries: + if not isinstance(entry, dict): + continue + page_id = str(entry.get("id", "")) + changes = entry.get("changes") + if not isinstance(changes, list): + continue + for change in changes: + if not isinstance(change, dict): + continue + value = change.get("value") + if not isinstance(value, dict): + continue + item = str(value.get("item", "")).strip().lower() + if item != "reaction": + continue + verb = str(value.get("verb", "")).strip().lower() + if verb != "add": + continue + from_obj = value.get("from") or {} + sender_id = ( + str(from_obj.get("id", "")) if isinstance(from_obj, dict) else "" + ) + sender_name = ( + str(from_obj.get("name", "")) if isinstance(from_obj, dict) else "" + ) + # Skip Page's own reactions to prevent self-loop. + if sender_id and sender_id == page_id: + FacebookAPI.logger.debug( + "iter_feed_reaction_events: skipping Page's own reaction " + "sender_id=%s == page_id=%s", + sender_id, + page_id, + ) + continue + reaction_type = ( + str(value.get("reaction_type", "") or "").strip().lower() + ) + post_id = str(value.get("post_id", "") or "").strip() + comment_id = str(value.get("comment_id", "") or "").strip() + ts = value.get("created_time") or value.get("timestamp") + try: + timestamp = int(ts) if ts is not None else 0 + except (TypeError, ValueError): + timestamp = 0 + out.append( + { + "sender_id": sender_id, + "sender_name": sender_name, + "page_id": page_id, + "post_id": post_id, + "comment_id": comment_id, + "reaction_type": reaction_type, + "message_type": "reaction", + "message": reaction_type, + "timestamp": timestamp, + "data": request, + "verb": verb, + } + ) + return out + def send_text_message(self, recipient_id: str, message: str) -> Dict: """Send text message to a Facebook user via Messenger.""" endpoint = f"{self.page_id}/messages" diff --git a/jvagent/action/facebook_action/facebook_comment_adapter.py b/jvagent/action/facebook_action/facebook_comment_adapter.py new file mode 100644 index 00000000..f7baca9f --- /dev/null +++ b/jvagent/action/facebook_action/facebook_comment_adapter.py @@ -0,0 +1,168 @@ +"""Facebook Page comment channel adapter for the response bus. + +Delivers agent replies as comment replies on Facebook Page posts using +the Graph API ``/{comment_id}/comments`` endpoint. Falls back to posting +a new comment on the original post (``/{post_id}/comments``) when only +``post_id`` is available, or to a Messenger DM when neither is present. +""" + +import asyncio +import logging +import re +from typing import Any, Dict + +from jvagent.action.response.channel_adapter import ChannelAdapter +from jvagent.action.response.message import ResponseMessage + +logger = logging.getLogger(__name__) + +FACEBOOK_COMMENT_MAX_LENGTH = 9000 + + +class FacebookCommentAdapter(ChannelAdapter): + """Deliver ``ResponseMessage`` content as Facebook Page comment replies. + + The adapter resolves the reply target from ``message.metadata``, which + carries the original webhook payload fields set in + ``messenger_webhook_helpers.create_feed_comment_walker``: + + - ``comment_id`` – reply to the specific comment (preferred) + - ``post_id`` – post a new comment on the original post (fallback) + - ``user_id`` – fall back to a Messenger DM if neither is available + """ + + def __init__(self, action: Any) -> None: + super().__init__(channel="facebook_comment") + self.action = action + self._user_locks: Dict[str, asyncio.Lock] = {} + + def _get_user_lock(self, user_id: str) -> asyncio.Lock: + if user_id not in self._user_locks: + if len(self._user_locks) >= 1000: + for key in list(self._user_locks.keys())[:100]: + del self._user_locks[key] + self._user_locks[user_id] = asyncio.Lock() + return self._user_locks[user_id] + + def _graph_failed(self, result: Any) -> bool: + if not isinstance(result, dict): + return True + return bool(result.get("error")) + + @staticmethod + def _strip_markdown(text: str) -> str: + """Reduce markdown to plain-text suitable for a public Facebook comment.""" + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + text = re.sub(r"\*(.+?)\*", r"\1", text) + text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) + text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text) + text = text.replace("
", "\n").replace("
", "\n") + text = text.replace("", "").replace("", "") + text = text.replace("", "").replace("", "") + text = text.replace("

", "\n").replace("

", "") + if len(text) > FACEBOOK_COMMENT_MAX_LENGTH: + text = text[: FACEBOOK_COMMENT_MAX_LENGTH - 3] + "..." + return text.strip() + + async def send(self, message: ResponseMessage) -> bool: + if not self.action or not self.action.is_configured(): + logger.debug("FacebookCommentAdapter: FacebookAction not configured") + return False + + comment_id = str(message.metadata.get("comment_id", "") or "").strip() + post_id = str(message.metadata.get("post_id", "") or "").strip() + user_id = str(message.user_id or "").strip() + + text = self._strip_markdown(str(message.content or "").strip()) + if not text: + logger.debug("FacebookCommentAdapter: empty message %s", message.id) + return False + + api = self.action.api() + + if comment_id: + logger.info("FacebookCommentAdapter: replying to comment %s", comment_id) + + def _reply() -> Any: + return api.reply_to_comment(comment_id, text) + + lock = self._get_user_lock(f"comment:{comment_id}") + async with lock: + try: + result = await asyncio.to_thread(_reply) + if self._graph_failed(result): + logger.error( + "FacebookCommentAdapter: reply_to_comment failed: %s", + result, + ) + return False + return True + except Exception as e: + logger.error( + "FacebookCommentAdapter: reply error for comment %s: %s", + comment_id, + e, + exc_info=True, + ) + return False + + if post_id: + logger.info("FacebookCommentAdapter: commenting on post %s", post_id) + + def _comment() -> Any: + return api.comment_on_post(post_id, text) + + lock = self._get_user_lock(f"post:{post_id}") + async with lock: + try: + result = await asyncio.to_thread(_comment) + if self._graph_failed(result): + logger.error( + "FacebookCommentAdapter: comment_on_post failed: %s", + result, + ) + return False + return True + except Exception as e: + logger.error( + "FacebookCommentAdapter: comment error for post %s: %s", + post_id, + e, + exc_info=True, + ) + return False + + if user_id: + logger.info( + "FacebookCommentAdapter: no comment_id/post_id, sending Messenger DM to %s", + user_id, + ) + + def _dm() -> Any: + return api.send_text_message(user_id, text) + + lock = self._get_user_lock(user_id) + async with lock: + try: + result = await asyncio.to_thread(_dm) + if self._graph_failed(result): + logger.error( + "FacebookCommentAdapter: Messenger DM fallback failed: %s", + result, + ) + return False + return True + except Exception as e: + logger.error( + "FacebookCommentAdapter: DM error for %s: %s", + user_id, + e, + exc_info=True, + ) + return False + + logger.error( + "FacebookCommentAdapter: no comment_id, post_id, or user_id on message %s", + message.id, + ) + return False diff --git a/jvagent/action/facebook_action/facebook_comment_filter.py b/jvagent/action/facebook_action/facebook_comment_filter.py new file mode 100644 index 00000000..2074430c --- /dev/null +++ b/jvagent/action/facebook_action/facebook_comment_filter.py @@ -0,0 +1,54 @@ +"""Facebook Page comment channel filter for the response bus. + +Strips markdown and HTML formatting that is inappropriate for public +Facebook comments (plain text only), and truncates to the Facebook +comment length limit. +""" + +import logging +import re +from typing import List, Optional + +from jvagent.action.response.channel_filter import ChannelFilter +from jvagent.action.response.message import ResponseMessage + +logger = logging.getLogger(__name__) + +FACEBOOK_COMMENT_MAX_LENGTH = 9000 + + +class FacebookCommentFilter(ChannelFilter): + """Transform markdown/HTML to plain text suitable for Facebook Page comments. + + Facebook comments are plain text — no markdown, no HTML. This filter + strips common formatting artifacts left by the ReplyAction voice. + """ + + def __init__( + self, channels: Optional[List[str]] = None, priority: int = 100 + ) -> None: + if channels is None: + channels = ["facebook_comment"] + super().__init__(channels=channels, priority=priority) + + async def filter(self, message: ResponseMessage) -> None: + if not message.content: + return + text = str(message.content) + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + text = re.sub(r"\*(.+?)\*", r"\1", text) + text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) + text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text) + text = ( + text.replace("
", "\n") + .replace("
", "\n") + .replace("", "") + .replace("", "") + .replace("", "") + .replace("", "") + .replace("

", "\n") + .replace("

", "") + ) + if len(text) > FACEBOOK_COMMENT_MAX_LENGTH: + text = text[: FACEBOOK_COMMENT_MAX_LENGTH - 3] + "..." + message.content = text.strip() diff --git a/jvagent/action/facebook_action/facebook_reaction_adapter.py b/jvagent/action/facebook_action/facebook_reaction_adapter.py new file mode 100644 index 00000000..99c113fe --- /dev/null +++ b/jvagent/action/facebook_action/facebook_reaction_adapter.py @@ -0,0 +1,38 @@ +"""Facebook Page reaction channel adapter for the response bus. + +Awareness-only adapter: reactions are ingested into the agent pipeline for +context (e.g. moderation triggers), but the agent does **not** auto-reply +or auto-react. ``send()`` logs and returns ``True`` without any Graph API +call. Custom tools/actions can inspect ``message.metadata`` for +``reaction_type``, ``post_id``, ``comment_id`` etc. if they want to act. +""" + +import logging +from typing import Any + +from jvagent.action.response.channel_adapter import ChannelAdapter +from jvagent.action.response.message import ResponseMessage + +logger = logging.getLogger(__name__) + + +class FacebookReactionAdapter(ChannelAdapter): + """No-op adapter for ``channel="facebook_reaction"``. + + The agent receives reaction events for awareness but does not deliver + any outbound response. ``send()`` silently succeeds so the response + bus considers the message delivered. + """ + + def __init__(self, action: Any = None) -> None: + super().__init__(channel="facebook_reaction") + self.action = action + + async def send(self, message: ResponseMessage) -> bool: + logger.debug( + "FacebookReactionAdapter: awareness-only, not replying to reaction " + "(message_id=%s, reaction_type=%s)", + message.id, + message.metadata.get("reaction_type", ""), + ) + return True diff --git a/jvagent/action/facebook_action/messenger_webhook_helpers.py b/jvagent/action/facebook_action/messenger_webhook_helpers.py index 57ca05bd..3b1ac18a 100644 --- a/jvagent/action/facebook_action/messenger_webhook_helpers.py +++ b/jvagent/action/facebook_action/messenger_webhook_helpers.py @@ -549,3 +549,201 @@ async def process_messenger_interaction_async( logger.error( "Error in messenger interaction for %s: %s", sender, e, exc_info=True ) + + +# --------------------------------------------------------------------------- +# Feed / comment webhook helpers (Facebook Page comment events) +# --------------------------------------------------------------------------- + +FEED_COMMENT_UTTERANCE_MAX = 2000 + +logger_feed = logging.getLogger(f"{__name__}.feed_comment") + + +async def create_feed_comment_walker( + agent_id: str, + utterance: str, + sender: str, + data_dict: Dict[str, Any], + sender_name: Optional[str] = None, +) -> Optional[InteractWalker]: + """Create an InteractWalker for a Facebook Page comment event. + + Uses ``channel="facebook_comment"`` so the channel_gate can distinguish + comments from Messenger DMs. + + Args: + agent_id: Agent ID to interact with. + utterance: Comment text (possibly empty if media-only). + sender: Facebook user ID of the commenter (not PSID — may be + a global user ID for Page commenters). + data_dict: Additional data dict (includes ``feed_payload``). + sender_name: Display name of the commenter. + """ + try: + convo_obj = await get_conversation_with_lock(sender) + + if convo_obj and getattr(convo_obj, "session_id", None): + return InteractWalker( + agent_id=agent_id, + utterance=utterance, + channel="facebook_comment", + data=data_dict, + session_id=convo_obj.session_id, + user_name=sender_name, + stream=False, + ) + return InteractWalker( + agent_id=agent_id, + utterance=utterance, + channel="facebook_comment", + data=data_dict, + user_id=sender, + user_name=sender_name, + stream=False, + ) + except Exception as e: + logger_feed.error("Error creating feed-comment walker for %s: %s", sender, e) + return None + + +async def process_feed_comment_interaction_async( + utterance: str, + sender: str, + agent_id: str, + agent: Any, + data_dict: Dict[str, Any], + sender_name: Optional[str] = None, +) -> None: + """Background task: adapter registration, walker spawn, finalize for a feed comment.""" + fb_action: Any = None + try: + fb_action = await agent.get_action_by_type("FacebookAction") + if fb_action: + await fb_action.ensure_adapter_registered() + except Exception as e: + logger_feed.warning( + "Feed-comment adapter ensure failed for agent %s: %s", agent_id, e + ) + + try: + walker = await create_feed_comment_walker( + agent_id, utterance, sender, data_dict, sender_name=sender_name + ) + if not walker: + return + await walker.spawn(agent) + await finalize_interaction_from_webhook(walker, agent_id, sender) + except DatabaseError: + raise + except Exception as e: + logger_feed.error( + "Error in feed-comment interaction for %s: %s", sender, e, exc_info=True + ) + + +# --------------------------------------------------------------------------- +# Feed / reaction webhook helpers (Facebook Page reaction events) +# --------------------------------------------------------------------------- + +REACTION_TYPE_LABELS: Dict[str, str] = { + "like": "👍 like", + "love": "❤️ love", + "wow": "😮 wow", + "haha": "😂 haha", + "sorry": "😢 sorry", + "anger": "😡 anger", +} + +FEED_REACTION_UTTERANCE_MAX = 500 + +logger_reaction = logging.getLogger(f"{__name__}.feed_reaction") + + +def _synthesize_reaction_utterance( + reaction_type: str, + sender_name: str, + comment_id: str, + post_id: str, +) -> str: + """Build a short utterance string that describes the reaction event.""" + label = REACTION_TYPE_LABELS.get(reaction_type, reaction_type) + target = "a comment" if comment_id else "a post" + name = sender_name or "Someone" + return f"[REACTION] {name} reacted with {label} on {target}." + + +async def create_feed_reaction_walker( + agent_id: str, + utterance: str, + sender: str, + data_dict: Dict[str, Any], + sender_name: Optional[str] = None, +) -> Optional[InteractWalker]: + """Create an InteractWalker for a Facebook Page reaction event. + + Uses ``channel="facebook_reaction"`` so the channel_gate and + FacebookReactionAdapter can distinguish reactions from comments and DMs. + """ + try: + convo_obj = await get_conversation_with_lock(sender) + + if convo_obj and getattr(convo_obj, "session_id", None): + return InteractWalker( + agent_id=agent_id, + utterance=utterance, + channel="facebook_reaction", + data=data_dict, + session_id=convo_obj.session_id, + user_name=sender_name, + stream=False, + ) + return InteractWalker( + agent_id=agent_id, + utterance=utterance, + channel="facebook_reaction", + data=data_dict, + user_id=sender, + user_name=sender_name, + stream=False, + ) + except Exception as e: + logger_reaction.error( + "Error creating feed-reaction walker for %s: %s", sender, e + ) + return None + + +async def process_feed_reaction_interaction_async( + utterance: str, + sender: str, + agent_id: str, + agent: Any, + data_dict: Dict[str, Any], + sender_name: Optional[str] = None, +) -> None: + """Background task: adapter registration, walker spawn, finalize for a feed reaction.""" + fb_action: Any = None + try: + fb_action = await agent.get_action_by_type("FacebookAction") + if fb_action: + await fb_action.ensure_adapter_registered() + except Exception as e: + logger_reaction.warning( + "Feed-reaction adapter ensure failed for agent %s: %s", agent_id, e + ) + + try: + walker = await create_feed_reaction_walker( + agent_id, utterance, sender, data_dict, sender_name=sender_name + ) + if not walker: + return + await walker.spawn(agent) + await finalize_interaction_from_webhook(walker, agent_id, sender) + except DatabaseError: + raise + except Exception as e: + logger_reaction.error( + "Error in feed-reaction interaction for %s: %s", sender, e, exc_info=True + ) From a1c6fb31e83b99a0b5f4c5e05ccec70dfe877262 Mon Sep 17 00:00:00 2001 From: Tharick Jairam Date: Fri, 7 Aug 2026 15:46:59 -0400 Subject: [PATCH 2/3] Remove noisy node_to_result content-length warning. Co-authored-by: Cursor --- jvagent/action/pageindex/models.py | 17 ----------------- tests/action/pageindex/test_pageindex.py | 8 +++----- 2 files changed, 3 insertions(+), 22 deletions(-) diff --git a/jvagent/action/pageindex/models.py b/jvagent/action/pageindex/models.py index 759ae04b..285be7f2 100644 --- a/jvagent/action/pageindex/models.py +++ b/jvagent/action/pageindex/models.py @@ -4,7 +4,6 @@ for graph-based persistence of document structure. """ -import logging from copy import deepcopy from typing import Any, Dict, List, Optional @@ -13,10 +12,6 @@ from .config import get_pageindex_retrieval_excerpt_source -_logger = logging.getLogger(__name__) - -_MAX_CONTENT_CHARS = 2000 - class DocumentRootNode(Node): """Root node for a document; links to top-level section nodes. @@ -157,18 +152,6 @@ def node_to_result( or "" ) - _CONTENT_WARN_THRESHOLD = _MAX_CONTENT_CHARS - if body_val and len(body_val) > _CONTENT_WARN_THRESHOLD: - _logger.warning( - "node_to_result: %s for node %s (%s) exceeds %d chars (%d). " - "Consider increasing observation_max_chars if the full content is needed.", - body_key, - node.node_id, - node.title[:80] if node.title else "", - _CONTENT_WARN_THRESHOLD, - len(body_val), - ) - # Citation/identity fields first so orchestrator middle-elision keeps them. # Public API / search rows use start_index/end_index (original field names). # pageindex__search remaps to start_page/end_page only when dumping into diff --git a/tests/action/pageindex/test_pageindex.py b/tests/action/pageindex/test_pageindex.py index c375c015..6c8753df 100644 --- a/tests/action/pageindex/test_pageindex.py +++ b/tests/action/pageindex/test_pageindex.py @@ -499,13 +499,11 @@ def test_node_to_result_excerpt_source_explicit(): def test_node_to_result_citation_fields_first_and_bodies_capped(): """Search hits put citation keys first; omit text/physical_index/enabled/hierarchy.""" - from jvagent.action.pageindex.models import _MAX_CONTENT_CHARS - n = DocumentNode() n.doc_name = "manual.pdf" n.title = "Section" - n.text = "T" * (_MAX_CONTENT_CHARS + 500) - n.summary = "S" * (_MAX_CONTENT_CHARS + 500) + n.text = "T" * 2500 + n.summary = "S" * 2500 n.start_index = 3 n.end_index = 5 n.structure = "1.2" @@ -524,7 +522,7 @@ def test_node_to_result_citation_fields_first_and_bodies_capped(): assert result["doc_name"] == "manual.pdf" assert result["start_index"] == 3 assert result["end_index"] == 5 - assert result["summary"] == "S" * (_MAX_CONTENT_CHARS + 500) + assert result["summary"] == "S" * 2500 def test_node_to_result_keeps_index_keys_no_hierarchy(): From 3b9109badef21ece3fc99dc00f841d8a33c1a940 Mon Sep 17 00:00:00 2001 From: Eldon Marks Date: Sat, 8 Aug 2026 08:05:37 -0400 Subject: [PATCH 3/3] fix(facebook): stop feed webhooks from posting duplicate public comments Review follow-ups on the comments/reactions work. Two of these compound into the same user-visible failure. Meta delivers webhooks at-least-once and retries for days. The Messenger path guards that on `mid` via remember_meta_wamid; the feed paths did not, so a redelivery posted a SECOND public reply under the same comment. Both now dedupe through the same helper. Comments key on the comment_id, which is stable and cannot be an edit -- only verb == "add" reaches here. Reactions carry no id, so the key is composed, and it includes the event timestamp on purpose: a retry repeats the timestamp, while a user who removes and re-adds a reaction produces a new one and is still heard. Keying on (post, comment, sender, type) alone would have silently eaten the second case. The retries were partly self-inflicted. Both feed paths awaited a full orchestrator turn inline in the webhook handler, holding the response open while the model and tools ran, so Meta timed the delivery out and resent it. They now dispatch via create_task with the same inline fallback the Messenger path uses for runtimes that cannot background. A comment longer than FEED_COMMENT_UTTERANCE_MAX was dropped with `continue` and no log -- the commenter got no reply and nothing explained why, while the reaction path 60 lines below truncated. It now truncates and logs. The markdown-to-plain-text transformation existed twice, once in the filter and once in the adapter, along with two independent declarations of FACEBOOK_COMMENT_MAX_LENGTH. Both now call facebook_comment_text, so a fix to one cannot leave the other emitting raw markdown into a public comment. Links keep their destination there -- `[label](url)` becomes `label (url)` -- because a comment reader cannot hover or recover a URL that was thrown away. Adds the first tests for this feature. TestEndpointWiring asserts on source structure rather than behaviour, and says so: driving the webhook needs a live graph, an agent and a model, so it guards that the dedup and the backgrounding are present, while the dedup keys themselves are exercised through the real helper. All three fixes were mutation-checked -- reverting each one turns the suite red. --- jvagent/action/facebook_action/endpoints.py | 95 +++++++++++++--- .../facebook_comment_adapter.py | 23 ++-- .../facebook_comment_filter.py | 28 +---- .../facebook_action/facebook_comment_text.py | 62 ++++++++++ .../facebook_action/test_feed_comment_text.py | 65 +++++++++++ .../test_feed_webhook_delivery.py | 107 ++++++++++++++++++ 6 files changed, 328 insertions(+), 52 deletions(-) create mode 100644 jvagent/action/facebook_action/facebook_comment_text.py create mode 100644 tests/action/facebook_action/test_feed_comment_text.py create mode 100644 tests/action/facebook_action/test_feed_webhook_delivery.py diff --git a/jvagent/action/facebook_action/endpoints.py b/jvagent/action/facebook_action/endpoints.py index d3288943..a8d7894d 100644 --- a/jvagent/action/facebook_action/endpoints.py +++ b/jvagent/action/facebook_action/endpoints.py @@ -774,12 +774,30 @@ async def handle_merged_messenger_event(merged_event: Dict[str, Any]) -> None: if not comment_text: continue if len(comment_text) > FEED_COMMENT_UTTERANCE_MAX: - continue + # Truncate rather than drop, matching the reaction path below. + # Dropping left a commenter with no reply and no log line to + # explain it, and Facebook comments run far longer than this cap. + logger.info( + "Feed comment %s truncated for the agent turn (%d > %d chars)", + str(comment_event.get("comment_id") or "?"), + len(comment_text), + FEED_COMMENT_UTTERANCE_MAX, + ) + comment_text = comment_text[: FEED_COMMENT_UTTERANCE_MAX - 3] + "..." comment_id = str(comment_event.get("comment_id") or "").strip() post_id = str(comment_event.get("post_id") or "").strip() sender_name = str(comment_event.get("sender_name") or "").strip() or None + # Meta delivers at-least-once and retries. Without this a redelivery + # posts a SECOND public reply under the same comment — the same guard + # the Messenger path applies to `mid` above. comment_id is stable per + # comment, and only verb == "add" reaches here, so an edit cannot + # masquerade as a new comment. + if comment_id and not remember_meta_wamid(f"fbcomment:{comment_id}"): + logger.debug("Facebook duplicate comment ignored: %s", comment_id) + continue + feed_data_dict: Dict[str, Any] = { "feed_payload": comment_event, "facebook_comment": True, @@ -805,14 +823,31 @@ async def handle_merged_messenger_event(merged_event: Dict[str, Any]) -> None: ) continue - await process_feed_comment_interaction_async( - comment_text, - sender, - agent_id, - agent, - feed_data_dict, - sender_name=sender_name, + # Background the turn like the Messenger path: awaiting a full + # orchestrator turn here holds the webhook response open, Meta times + # the delivery out and retries, and the retry arrives as another + # inbound comment. create_task returns None where the runtime cannot + # background (serverless), where awaiting inline is the only option. + comment_task = await create_task( + process_feed_comment_interaction_async( + comment_text, + sender, + agent_id, + agent, + feed_data_dict, + sender_name=sender_name, + ), + name=f"facebook_comment_interaction_{sender}", ) + if comment_task is None: + await process_feed_comment_interaction_async( + comment_text, + sender, + agent_id, + agent, + feed_data_dict, + sender_name=sender_name, + ) # Process feed/reaction events (Facebook Page reactions) for reaction_event in feed_reaction_events: @@ -841,6 +876,24 @@ async def handle_merged_messenger_event(merged_event: Dict[str, Any]) -> None: if len(utterance) > FEED_REACTION_UTTERANCE_MAX: utterance = utterance[: FEED_REACTION_UTTERANCE_MAX - 3] + "..." + # Same at-least-once delivery guard as comments. A reaction carries no + # id of its own, so the key is composed — and it INCLUDES the event + # timestamp on purpose: a retry repeats the timestamp, while a user who + # removes and re-adds a reaction produces a new one and is still heard. + reaction_key = ":".join( + ( + "fbreaction", + post_id, + comment_id, + sender, + reaction_type, + str(reaction_event.get("timestamp") or ""), + ) + ) + if not remember_meta_wamid(reaction_key): + logger.debug("Facebook duplicate reaction ignored: %s", reaction_key) + continue + reaction_data_dict: Dict[str, Any] = { "feed_payload": reaction_event, "facebook_reaction": True, @@ -867,14 +920,26 @@ async def handle_merged_messenger_event(merged_event: Dict[str, Any]) -> None: ) continue - await process_feed_reaction_interaction_async( - utterance, - sender, - agent_id, - agent, - reaction_data_dict, - sender_name=sender_name, + reaction_task = await create_task( + process_feed_reaction_interaction_async( + utterance, + sender, + agent_id, + agent, + reaction_data_dict, + sender_name=sender_name, + ), + name=f"facebook_reaction_interaction_{sender}", ) + if reaction_task is None: + await process_feed_reaction_interaction_async( + utterance, + sender, + agent_id, + agent, + reaction_data_dict, + sender_name=sender_name, + ) return {"status": "received"} diff --git a/jvagent/action/facebook_action/facebook_comment_adapter.py b/jvagent/action/facebook_action/facebook_comment_adapter.py index f7baca9f..1a3976ee 100644 --- a/jvagent/action/facebook_action/facebook_comment_adapter.py +++ b/jvagent/action/facebook_action/facebook_comment_adapter.py @@ -8,15 +8,14 @@ import asyncio import logging -import re from typing import Any, Dict from jvagent.action.response.channel_adapter import ChannelAdapter from jvagent.action.response.message import ResponseMessage -logger = logging.getLogger(__name__) +from .facebook_comment_text import to_facebook_comment_text -FACEBOOK_COMMENT_MAX_LENGTH = 9000 +logger = logging.getLogger(__name__) class FacebookCommentAdapter(ChannelAdapter): @@ -51,18 +50,12 @@ def _graph_failed(self, result: Any) -> bool: @staticmethod def _strip_markdown(text: str) -> str: - """Reduce markdown to plain-text suitable for a public Facebook comment.""" - text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) - text = re.sub(r"\*(.+?)\*", r"\1", text) - text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) - text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text) - text = text.replace("
", "\n").replace("
", "\n") - text = text.replace("", "").replace("", "") - text = text.replace("", "").replace("", "") - text = text.replace("

", "\n").replace("

", "") - if len(text) > FACEBOOK_COMMENT_MAX_LENGTH: - text = text[: FACEBOOK_COMMENT_MAX_LENGTH - 3] + "..." - return text.strip() + """Reduce markdown to plain text for a public Facebook comment. + + Delegates to the shared helper so this and ``FacebookCommentFilter`` + cannot drift apart. + """ + return to_facebook_comment_text(text) async def send(self, message: ResponseMessage) -> bool: if not self.action or not self.action.is_configured(): diff --git a/jvagent/action/facebook_action/facebook_comment_filter.py b/jvagent/action/facebook_action/facebook_comment_filter.py index 2074430c..c36d3c98 100644 --- a/jvagent/action/facebook_action/facebook_comment_filter.py +++ b/jvagent/action/facebook_action/facebook_comment_filter.py @@ -6,22 +6,23 @@ """ import logging -import re from typing import List, Optional from jvagent.action.response.channel_filter import ChannelFilter from jvagent.action.response.message import ResponseMessage -logger = logging.getLogger(__name__) +from .facebook_comment_text import to_facebook_comment_text -FACEBOOK_COMMENT_MAX_LENGTH = 9000 +logger = logging.getLogger(__name__) class FacebookCommentFilter(ChannelFilter): """Transform markdown/HTML to plain text suitable for Facebook Page comments. Facebook comments are plain text — no markdown, no HTML. This filter - strips common formatting artifacts left by the ReplyAction voice. + strips common formatting artifacts left by the ReplyAction voice. The + transformation lives in ``facebook_comment_text`` so the adapter applies + exactly the same one. """ def __init__( @@ -34,21 +35,4 @@ def __init__( async def filter(self, message: ResponseMessage) -> None: if not message.content: return - text = str(message.content) - text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) - text = re.sub(r"\*(.+?)\*", r"\1", text) - text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) - text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text) - text = ( - text.replace("
", "\n") - .replace("
", "\n") - .replace("", "") - .replace("", "") - .replace("", "") - .replace("", "") - .replace("

", "\n") - .replace("

", "") - ) - if len(text) > FACEBOOK_COMMENT_MAX_LENGTH: - text = text[: FACEBOOK_COMMENT_MAX_LENGTH - 3] + "..." - message.content = text.strip() + message.content = to_facebook_comment_text(str(message.content)) diff --git a/jvagent/action/facebook_action/facebook_comment_text.py b/jvagent/action/facebook_action/facebook_comment_text.py new file mode 100644 index 00000000..feceb057 --- /dev/null +++ b/jvagent/action/facebook_action/facebook_comment_text.py @@ -0,0 +1,62 @@ +"""Plain-text shaping for Facebook Page comments. + +Facebook comments are plain text — no markdown, no HTML. Both the channel +filter (bus pipeline) and the adapter (delivery, for content that never went +through the filter) need the same transformation, and having it written twice +means the two drift: a fix to one silently leaves the other emitting raw +markdown into a public comment. +""" + +from __future__ import annotations + +import re + +# Facebook's documented comment ceiling. Defined once so the filter and the +# adapter cannot disagree about where to cut. +FACEBOOK_COMMENT_MAX_LENGTH = 9000 + +_BOLD = re.compile(r"\*\*(.+?)\*\*") +_ITALIC = re.compile(r"\*(.+?)\*") +_HEADING = re.compile(r"^#{1,6}\s+", re.MULTILINE) +_LINK = re.compile(r"\[([^\]]+)\]\(([^)]+)\)") + +_HTML_REPLACEMENTS = ( + ("
", "\n"), + ("
", "\n"), + ("", ""), + ("", ""), + ("", ""), + ("", ""), + ("

", "\n"), + ("

", ""), +) + + +def _link_to_text(match: "re.Match[str]") -> str: + """``[label](url)`` -> ``label (url)``, or just ``label`` when redundant. + + A comment reader has no hover target and no way to recover a dropped URL, + so the destination is kept inline rather than discarded. When the label + already *is* the URL, repeating it would only add noise. + """ + label = match.group(1).strip() + url = match.group(2).strip() + if not url or label == url: + return label + return f"{label} ({url})" + + +def to_facebook_comment_text(text: str) -> str: + """Reduce markdown/HTML to plain text and bound it to the comment limit.""" + if not text: + return "" + out = _BOLD.sub(r"\1", text) + out = _ITALIC.sub(r"\1", out) + out = _HEADING.sub("", out) + out = _LINK.sub(_link_to_text, out) + for needle, replacement in _HTML_REPLACEMENTS: + out = out.replace(needle, replacement) + out = out.strip() + if len(out) > FACEBOOK_COMMENT_MAX_LENGTH: + out = out[: FACEBOOK_COMMENT_MAX_LENGTH - 3].rstrip() + "..." + return out diff --git a/tests/action/facebook_action/test_feed_comment_text.py b/tests/action/facebook_action/test_feed_comment_text.py new file mode 100644 index 00000000..50fe18af --- /dev/null +++ b/tests/action/facebook_action/test_feed_comment_text.py @@ -0,0 +1,65 @@ +"""Plain-text shaping for Facebook Page comments. + +The filter (bus pipeline) and the adapter (delivery) must apply the *same* +transformation — two copies drift, and the failure is raw markdown in a public +comment under a brand's post. +""" + +from __future__ import annotations + +from jvagent.action.facebook_action.facebook_comment_adapter import ( + FacebookCommentAdapter, +) +from jvagent.action.facebook_action.facebook_comment_text import ( + FACEBOOK_COMMENT_MAX_LENGTH, + to_facebook_comment_text, +) + + +class TestMarkdownStripping: + def test_bold_italic_and_headings(self) -> None: + out = to_facebook_comment_text("## Title\n**bold** and *italic*") + assert out == "Title\nbold and italic" + + def test_html_tags_become_plain_text(self) -> None: + out = to_facebook_comment_text("a
bcd") + assert out == "a\nbcd" + + def test_link_keeps_its_url(self) -> None: + """A comment reader cannot hover or recover a dropped URL.""" + out = to_facebook_comment_text("See [our pricing](https://x.test/pricing).") + assert out == "See our pricing (https://x.test/pricing)." + + def test_link_whose_label_is_the_url_is_not_repeated(self) -> None: + out = to_facebook_comment_text("[https://x.test](https://x.test)") + assert out == "https://x.test" + + def test_truncates_to_the_facebook_limit(self) -> None: + out = to_facebook_comment_text("x" * (FACEBOOK_COMMENT_MAX_LENGTH + 500)) + assert len(out) <= FACEBOOK_COMMENT_MAX_LENGTH + assert out.endswith("...") + + def test_empty_input_is_safe(self) -> None: + assert to_facebook_comment_text("") == "" + + +class TestSingleImplementation: + def test_adapter_and_helper_agree(self) -> None: + """The adapter must not carry its own copy of the transformation.""" + sample = "**Hi** — see [docs](https://x.test/d)
now" + assert FacebookCommentAdapter._strip_markdown(sample) == ( + to_facebook_comment_text(sample) + ) + + async def test_filter_and_helper_agree(self) -> None: + from jvagent.action.facebook_action.facebook_comment_filter import ( + FacebookCommentFilter, + ) + from jvagent.action.response.message import ResponseMessage + + sample = "## Heading\n**bold** [link](https://x.test)" + message = ResponseMessage( + session_id="s", user_id="u", content=sample, channel="facebook_comment" + ) + await FacebookCommentFilter().filter(message) + assert message.content == to_facebook_comment_text(sample) diff --git a/tests/action/facebook_action/test_feed_webhook_delivery.py b/tests/action/facebook_action/test_feed_webhook_delivery.py new file mode 100644 index 00000000..58b8883e --- /dev/null +++ b/tests/action/facebook_action/test_feed_webhook_delivery.py @@ -0,0 +1,107 @@ +"""At-least-once delivery handling for Facebook Page feed webhooks. + +Meta delivers webhooks at-least-once and retries for days. Two properties keep +that from turning into duplicate **public** comments under a brand's post, and +both are invisible until it happens in production: + +- a redelivered event is recognised and dropped (the Messenger path already + does this for ``mid``) +- the agent turn is backgrounded, so the webhook answers immediately instead of + holding the connection open long enough for Meta to time out and retry + +The second causes the first: a slow turn *creates* the redelivery. +""" + +from __future__ import annotations + +import pytest + +from jvagent.action.utils.meta_webhook_dedup import remember_meta_wamid + + +@pytest.fixture(autouse=True) +def _isolate_dedup_state(monkeypatch: pytest.MonkeyPatch) -> None: + """Dedup state is process-global; keep cases independent.""" + from jvagent.action.utils import meta_webhook_dedup as dedup + + monkeypatch.setattr(dedup, "_seen_wamids", type(dedup._seen_wamids)()) + monkeypatch.setattr(dedup, "_redis_client", None) + monkeypatch.setattr(dedup, "_redis_init_attempted", True) + + +class TestDedupKeys: + """The keys the endpoint builds, exercised through the real dedup helper.""" + + def test_same_comment_id_is_seen_once(self) -> None: + key = "fbcomment:c_1" + assert remember_meta_wamid(key) is True + assert remember_meta_wamid(key) is False + + def test_distinct_comments_both_pass(self) -> None: + assert remember_meta_wamid("fbcomment:c_1") is True + assert remember_meta_wamid("fbcomment:c_2") is True + + def test_reaction_retry_is_suppressed(self) -> None: + """Identical event incl. timestamp = a redelivery.""" + key = ":".join(("fbreaction", "p_1", "c_1", "u_1", "LIKE", "1700000000")) + assert remember_meta_wamid(key) is True + assert remember_meta_wamid(key) is False + + def test_reacting_again_later_is_still_heard(self) -> None: + """Remove-then-re-add produces a new timestamp and must not be eaten. + + This is why the reaction key includes the timestamp rather than being + keyed on (post, comment, sender, type) alone. + """ + base = ("fbreaction", "p_1", "c_1", "u_1", "LIKE") + assert remember_meta_wamid(":".join((*base, "1700000000"))) is True + assert remember_meta_wamid(":".join((*base, "1700000900"))) is True + + def test_comment_and_reaction_namespaces_do_not_collide(self) -> None: + assert remember_meta_wamid("fbcomment:x") is True + assert remember_meta_wamid("fbreaction:x") is True + + +class TestEndpointWiring: + """Guard the shape of the handler itself. + + These assert on source structure rather than behaviour because driving the + full webhook needs a live graph, an agent and a model; the properties worth + protecting here are "the guard is present" and "the turn is not awaited + inline", both of which a refactor can silently undo. + """ + + @staticmethod + def _source() -> str: + from pathlib import Path + + import jvagent.action.facebook_action.endpoints as endpoints_module + + return Path(endpoints_module.__file__).read_text(encoding="utf-8") + + def test_feed_paths_dedupe_before_dispatch(self) -> None: + source = self._source() + assert "fbcomment:" in source, "comment path lost its dedup key" + assert "fbreaction" in source, "reaction path lost its dedup key" + + def test_feed_turns_are_backgrounded(self) -> None: + """A bare ``await process_feed_*`` blocks the webhook response. + + Asserted via the task names, which exist only on the ``create_task`` + call — matching the call's own formatting would break on a reformat. + """ + source = self._source() + assert "create_task" in source + for task_name in ( + "facebook_comment_interaction_", + "facebook_reaction_interaction_", + ): + assert task_name in source, ( + f"expected a backgrounded task named {task_name!r} so the " + "webhook answers before Meta times out and retries" + ) + + def test_long_comment_is_truncated_not_dropped(self) -> None: + """Dropping left the commenter with no reply and no log line.""" + source = self._source() + assert "comment_text[: FEED_COMMENT_UTTERANCE_MAX - 3]" in source