diff --git a/jvagent/action/facebook_action/endpoints.py b/jvagent/action/facebook_action/endpoints.py
index 2af99e80..a8d7894d 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,191 @@ 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:
+ # 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,
+ "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
+
+ # 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:
+ 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] + "..."
+
+ # 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,
+ "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
+
+ 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"}
@@ -805,3 +998,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..1a3976ee
--- /dev/null
+++ b/jvagent/action/facebook_action/facebook_comment_adapter.py
@@ -0,0 +1,161 @@
+"""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
+from typing import Any, Dict
+
+from jvagent.action.response.channel_adapter import ChannelAdapter
+from jvagent.action.response.message import ResponseMessage
+
+from .facebook_comment_text import to_facebook_comment_text
+
+logger = logging.getLogger(__name__)
+
+
+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 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():
+ 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..c36d3c98
--- /dev/null
+++ b/jvagent/action/facebook_action/facebook_comment_filter.py
@@ -0,0 +1,38 @@
+"""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
+from typing import List, Optional
+
+from jvagent.action.response.channel_filter import ChannelFilter
+from jvagent.action.response.message import ResponseMessage
+
+from .facebook_comment_text import to_facebook_comment_text
+
+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. The
+ transformation lives in ``facebook_comment_text`` so the adapter applies
+ exactly the same one.
+ """
+
+ 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
+ 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/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
+ )
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