diff --git a/CHANGELOG.md b/CHANGELOG.md
index ed05a822..17019164 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,11 @@ and this project adheres to [PEP 440](https://peps.python.org/pep-0440/) /
 
 ### Fixed
 
+- **`voice.closers` no longer strips trailing questions.** Scrub peel skipped
+  sentences ending in `?` / `?!` so prompts like “Could you let me know if you
+  need a quote?” stay intact; dropped the broad `let me know if` closer pattern
+  that caused false positives.
+
 - **No WARNING when ambient core parameters are re-unioned.**
   `resolve_parameters` still drops duplicate inviolable floors, but skips the
   conflict log when the challenger is itself an ambient/core inviolable (pools
@@ -68,6 +73,25 @@ and this project adheres to [PEP 440](https://peps.python.org/pep-0440/) /
 
 ### Changed
 
+- **Lean PageIndex search hits.** `node_to_result` puts citation/identity fields
+  first (`doc_name`, pages, `node_id`, …) and omits bulk `text` /
+  `physical_index` / `enabled` by default so orchestrator middle-elision keeps
+  citations. Cap `content` and `summary`; request body text via
+  `include=["text"]` or `excerpt_source="text"`. `start_index` / `end_index`
+  are still returned alongside the new `start_page` / `end_page` aliases —
+  these rows are the public search endpoint's response, so dropping them would
+  break API clients for no budget saving. Coverage:
+  `tests/action/pageindex/test_pageindex.py`.
+
+- **Compact `pageindex__list` tool payload.** Returns `{count, documents}` with
+  truncated descriptions (drops `doc_url` / `root_id` / collection) and shrinks
+  further to stay under the orchestrator observation budget
+  (`DEFAULT_OBSERVATION_MAX_CHARS`). When even minimum-length descriptions do
+  not fit, whole entries are dropped and the payload reports `shown` +
+  `truncated: true` rather than silently returning a partial list. `summary`
+  defaults to true; `summary=false` may add `access` / `chunks` when they
+  still fit.
+
 - **Library skill `artifact_handler` is opt-in, not always-active.** Removed
   `always-active: true` so agents with `skills_source: both`/`library` no longer
   auto-discover and then hide the skill when `ArtifactHandlerInteractAction` is
diff --git a/jvagent/action/artifact_handler_interact_action/endpoints.py b/jvagent/action/artifact_handler_interact_action/endpoints.py
index 652b3fcb..6abcefa0 100644
--- a/jvagent/action/artifact_handler_interact_action/endpoints.py
+++ b/jvagent/action/artifact_handler_interact_action/endpoints.py
@@ -3,7 +3,8 @@
 jvforge POSTs to ``/api/artifact_handler_action/notify/{agent_id}`` with a
 ``process_document_url`` when an async ingest job finishes. The vault
 downloads the artifact, imports the pageindex_graph into PageIndex, then
-sends two WhatsApp messages: a ready notice (immediate) and an answer
+sends a proactive notification (WhatsApp or Messenger) with a ready notice
+and an optional answer.
 (background, using call_model if there's a pending question).
 
 On failure the endpoint returns 503 + Retry-After so jvforge retries the
@@ -328,6 +329,188 @@ async def _publish_whatsapp_message(
         return False
 
 
+async def _publish_messenger_message(
+    *,
+    agent: Any,
+    user_id: str,
+    session_id: str,
+    conversation_id: str,
+    content: str,
+    display_doc: str,
+    job_id: str,
+    answered: bool = False,
+) -> bool:
+    """Send a Facebook Messenger message via the registered FacebookAction.
+
+    Creates an interaction for record-keeping, sets the response, and sends
+    via ``FacebookAPI.send_text_message()`` on the live FacebookAction held by
+    the registered MessengerAdapter — never via ``response_bus.publish``, which
+    would append a duplicate reply onto the interaction.
+    """
+    memory = await agent.get_memory()
+    if not memory:
+        logger.warning("_publish_messenger_message: agent has no memory, cannot send")
+        return False
+
+    conversation = None
+    if conversation_id:
+        try:
+            from jvagent.memory.conversation import Conversation
+
+            conversation = await Conversation.get(conversation_id)
+        except Exception:
+            conversation = None
+
+    if conversation is None:
+        user = await memory.get_user(user_id, create_if_missing=False)
+        if not user:
+            logger.warning(
+                "_publish_messenger_message: user not found user_id=%s", user_id
+            )
+            return False
+        if session_id:
+            conversation = await user.get_conversation_by_session(session_id)
+        if conversation is None:
+            logger.warning(
+                "_publish_messenger_message: conversation not found "
+                "user_id=%s session_id=%s conversation_id=%s",
+                user_id,
+                session_id,
+                conversation_id,
+            )
+            return False
+
+    effective_session_id = (
+        session_id or str(getattr(conversation, "session_id", "") or "").strip() or ""
+    )
+    if not effective_session_id:
+        logger.warning(
+            "_publish_messenger_message: no effective session_id "
+            "user_id=%s conversation_id=%s",
+            user_id,
+            conversation_id,
+        )
+        return False
+
+    interaction = await conversation.add_interaction(
+        utterance="",
+        channel="messenger",
+        session_id=effective_session_id,
+    )
+    if not interaction:
+        logger.warning(
+            "_publish_messenger_message: add_interaction returned None "
+            "user_id=%s conversation_id=%s",
+            user_id,
+            conversation_id,
+        )
+        return False
+
+    interaction.add_parameter(
+        {
+            "is_proactive": True,
+            "job_id": job_id,
+            "doc_name": display_doc,
+            "ready": True,
+            "answered": answered,
+        },
+        "ArtifactHandlerInteractAction",
+    )
+
+    if content and content.strip():
+        interaction.set_response(content.strip())
+
+    await interaction.save()
+
+    # Use the already-registered live FacebookAction held by MessengerAdapter
+    # (startup-resolved Page token). Never call api() on a fresh find_one instance.
+    try:
+        response_bus = await agent.get_response_bus()
+    except Exception:
+        logger.warning(
+            "_publish_messenger_message: get_response_bus failed",
+            exc_info=True,
+        )
+        return False
+    if not response_bus:
+        logger.warning("_publish_messenger_message: no response bus")
+        return False
+
+    adapter = response_bus._channel_adapters.get("messenger")
+    if not adapter or not getattr(adapter, "_initialized", False):
+        facebook_action = await agent.get_action_by_type("FacebookAction")
+        if facebook_action is None:
+            logger.warning(
+                "_publish_messenger_message: FacebookAction not found on agent"
+            )
+            return False
+        try:
+            await facebook_action.ensure_page_access_token()
+            await facebook_action.ensure_adapter_registered()
+        except Exception:
+            logger.warning(
+                "_publish_messenger_message: ensure adapter/token failed",
+                exc_info=True,
+            )
+            return False
+        adapter = response_bus._channel_adapters.get("messenger")
+        if not adapter:
+            logger.warning(
+                "_publish_messenger_message: MessengerAdapter not registered"
+            )
+            return False
+
+    facebook_action = getattr(adapter, "action", None)
+    if facebook_action is None:
+        logger.warning(
+            "_publish_messenger_message: MessengerAdapter has no FacebookAction"
+        )
+        return False
+
+    try:
+        if not facebook_action.is_configured():
+            logger.warning("_publish_messenger_message: FacebookAction not configured")
+            return False
+    except Exception:
+        logger.warning(
+            "_publish_messenger_message: FacebookAction is_configured() failed",
+            exc_info=True,
+        )
+        return False
+
+    try:
+        api = facebook_action.api()
+    except Exception:
+        logger.warning(
+            "_publish_messenger_message: FacebookAction.api() failed on "
+            "registered action",
+            exc_info=True,
+        )
+        return False
+
+    try:
+        result = await asyncio.to_thread(api.send_text_message, user_id, content)
+        if isinstance(result, dict) and result.get("error"):
+            logger.error(
+                "_publish_messenger_message: send_text_message error for "
+                "user_id=%s: %s",
+                user_id,
+                result.get("error"),
+            )
+            return False
+        logger.info(
+            "_publish_messenger_message: sent to user_id=%s job_id=%s", user_id, job_id
+        )
+        return True
+    except Exception:
+        logger.error(
+            "_publish_messenger_message: send_text_message exception for user_id=%s",
+            user_id,
+            exc_info=True,
+        )
+        return False
+
+
 _PROCESSING_STATUSES = frozenset({"queued", "processing", "pending", "submitted"})
 
 _RETRY_AFTER_SECONDS = 30
@@ -448,7 +631,7 @@ def _canned_ready_message(
     doc_description: Optional[str] = None,
     pending_question: Optional[str] = None,
 ) -> str:
-    """Fallback WhatsApp notification (single message, never 'file').
+    """Fallback notification message (single message, never 'file').
 
     When a pending question exists: ready → remind question → invite answer
     follow-up (no LLM answer available in this fallback).
@@ -477,7 +660,7 @@ def _canned_ready_message_multi(
     doc_descriptions: Optional[Dict[str, str]] = None,
     pending_questions: Optional[Dict[str, str]] = None,
 ) -> str:
-    """Consolidated WhatsApp ready notice for multiple documents."""
+    """Consolidated ready notice for multiple documents."""
     if not display_docs:
         return "Your files are ready. Ask me anything about them."
     if len(display_docs) == 1:
@@ -582,7 +765,7 @@ async def _generate_ready_message(
 
     has_question = bool((utterance or "").strip())
     system_parts = [
-        "You write a single concise WhatsApp reply. Follow these rules exactly:",
+        "You write a single concise reply. Follow these rules exactly:",
         f"- Briefly state that the {kind} is ready (e.g. 'Your {type_word} is ready'). {name_guidance} Never call it a 'file'.",
     ]
     if has_question:
@@ -626,12 +809,10 @@ async def _generate_ready_message(
             f"\nSearch excerpts for doc_name={internal_doc_name!r}:\n{excerpts}"
         )
         user_parts.append(
-            "\nWrite one short WhatsApp message: ready → remind question → answer."
+            "\nWrite one short message: ready → remind question → answer."
         )
     else:
-        user_parts.append(
-            "\nNo pending question. Write one short WhatsApp ready notice."
-        )
+        user_parts.append("\nNo pending question. Write one short ready notice.")
     user_prompt = "\n".join(user_parts)
 
     try:
@@ -718,7 +899,7 @@ async def _generate_ready_message_multi(
 
     filenames_line = ", ".join(repr(d) for d in display_docs)
     system_parts = [
-        "You write natural WhatsApp replies. Follow these rules exactly:",
+        "You write natural replies. Follow these rules exactly:",
         f"- Always tell the user their {kinds_label} {'are' if is_plural else 'is'} ready. "
         f"Refer to each document by its type word (e.g. 'your PDF', 'your image') "
         f"unless the filename is clearly meaningful and descriptive — if a "
@@ -778,12 +959,10 @@ async def _generate_ready_message_multi(
         user_parts.append("Search excerpts:")
         user_parts.extend(search_parts)
         user_parts.append("")
-        user_parts.append(
-            "Write one WhatsApp message: ready → remind question(s) → answer(s)."
-        )
+        user_parts.append("Write one message: ready → remind question(s) → answer(s).")
     else:
         user_parts.append("")
-        user_parts.append("No pending questions. Write one WhatsApp ready notice.")
+        user_parts.append("No pending questions. Write one short ready notice.")
 
     user_prompt = "\n".join(user_parts)
 
@@ -836,7 +1015,7 @@ async def artifact_handler_notify(request: Request, agent_id: str):
         2. Require a known ``job_id`` in the reverse index (blocks replay/spam import).
         3. Download artifact from ``process_document_url`` and import into PageIndex.
         4. Mark the job as ``ready`` in conversation ``pending_ingest_jobs``.
-        5. For WhatsApp: send ready notice + optional answer.
+        5. For WhatsApp/Messenger: send ready notice + optional answer.
         6. Return 200 on success, 503 + Retry-After on failure (so jvforge retries).
     """
     import hmac
@@ -942,6 +1121,13 @@ async def artifact_handler_notify(request: Request, agent_id: str):
     session_id = str(entry.get("session_id") or "").strip()
     conversation_id = str(entry.get("conversation_id") or "").strip()
     channel = str(entry.get("channel") or "").strip().lower() or "default"
+    logger.info(
+        "artifact_handler_notify: job_id=%s channel=%s user_id=%s doc=%s",
+        job_id,
+        channel,
+        user_id,
+        doc_name,
+    )
     # Prefer PageIndex import name; fall back to vault job name normalized the
     # same way PageIndex does (strip_redundant_md_suffix).
     vault_doc_name = str(entry.get("doc_name") or doc_name or "").strip()
@@ -988,7 +1174,9 @@ async def artifact_handler_notify(request: Request, agent_id: str):
                                     )
         except Exception:
             pass
-    # ── Send WhatsApp notifications (ready notice + answer).
+    # ── Send proactive notifications.
+    # WhatsApp and Messenger get push messages; web/default relies on
+    # check_ingest_status polling (TODO: add web push in a future phase).
     if user_id and channel == "whatsapp":
         asyncio.create_task(
             _send_whatsapp_notifications(
@@ -1002,6 +1190,19 @@ async def artifact_handler_notify(request: Request, agent_id: str):
                 pending_question=pending_question,
             )
         )
+    elif user_id and channel == "messenger":
+        asyncio.create_task(
+            _send_messenger_notifications(
+                agent_id=agent_id,
+                job_id=job_id or "",
+                user_id=user_id,
+                session_id=session_id,
+                conversation_id=conversation_id,
+                internal_doc_name=internal_doc_name,
+                display_doc=display_doc,
+                pending_question=pending_question,
+            )
+        )
 
     # ── Mark notified + clear from jvforge reverse index.
     if action is not None and job_id:
@@ -1013,7 +1214,7 @@ async def artifact_handler_notify(request: Request, agent_id: str):
     return {
         "status": "imported",
         "job_id": job_id,
-        "notified": channel == "whatsapp" and bool(user_id),
+        "notified": channel in ("whatsapp", "messenger") and bool(user_id),
         "doc_name": imported_doc_name,
     }
 
@@ -1090,3 +1291,96 @@ async def _send_whatsapp_notifications(
             job_id,
             exc_info=True,
         )
+
+
+async def _send_messenger_notifications(
+    *,
+    agent_id: str,
+    job_id: str,
+    user_id: str,
+    session_id: str,
+    conversation_id: str,
+    internal_doc_name: str,
+    display_doc: str,
+    pending_question: str,
+) -> None:
+    """Send a single Messenger notification: ready notice, or ready + answer."""
+    logger.info(
+        "_send_messenger_notifications: starting agent_id=%s job_id=%s "
+        "user_id=%s doc=%s",
+        agent_id,
+        job_id,
+        user_id,
+        display_doc,
+    )
+    try:
+        from jvagent.core.agent import Agent
+
+        agent = await Agent.get(agent_id)
+        if agent is None:
+            logger.warning(
+                "_send_messenger_notifications: agent not found agent_id=%s",
+                agent_id,
+            )
+            return
+
+        action = await _resolve_action(agent_id)
+
+        single_entry = {
+            "internal_doc_name": internal_doc_name,
+            "display_doc": display_doc,
+            "pending_question": pending_question,
+        }
+        desc_lookup: Dict[str, str] = {}
+        try:
+            desc_lookup = await _doc_description_lookup(agent, [single_entry])
+        except Exception:
+            pass
+        doc_description = desc_lookup.get(internal_doc_name, "")
+
+        content: Optional[str] = None
+        answered = False
+
+        if pending_question and internal_doc_name and action is not None:
+            content = await _generate_ready_message(
+                agent=agent,
+                vault_action=action,
+                internal_doc_name=internal_doc_name,
+                display_doc=display_doc,
+                utterance=pending_question,
+                doc_description=doc_description or None,
+            )
+            if content:
+                answered = True
+
+        if not content:
+            content = _canned_ready_message(
+                display_doc,
+                doc_description=doc_description,
+                pending_question=pending_question or None,
+            )
+
+        logger.info(
+            "_send_messenger_notifications: publishing to user_id=%s "
+            "answered=%s content_len=%d",
+            user_id,
+            answered,
+            len(content) if content else 0,
+        )
+        await _publish_messenger_message(
+            agent=agent,
+            user_id=user_id,
+            session_id=session_id,
+            conversation_id=conversation_id,
+            content=content,
+            display_doc=display_doc,
+            job_id=job_id,
+            answered=answered,
+        )
+    except Exception:
+        logger.error(
+            "_send_messenger_notifications: unexpected error agent_id=%s job_id=%s",
+            agent_id,
+            job_id,
+            exc_info=True,
+        )
diff --git a/jvagent/action/facebook_action/endpoints.py b/jvagent/action/facebook_action/endpoints.py
index d988c222..2af99e80 100644
--- a/jvagent/action/facebook_action/endpoints.py
+++ b/jvagent/action/facebook_action/endpoints.py
@@ -614,7 +614,6 @@ async def _agent_and_facebook_action_for_messenger_webhook(
     methods=["GET"],
     webhook=True,
     auth=False,
-    webhook_auth="api_key",  # Validates API key from query param or header
     tags=["Facebook Action", "Messenger"],
     summary="Meta Messenger webhook: GET hub challenge (subscription verify)",
 )
@@ -635,7 +634,6 @@ async def messenger_interact_webhook_verify(request: Request, agent_id: str) ->
     methods=["POST"],
     webhook=True,
     auth=False,
-    webhook_auth="api_key",  # Validates API key from query param or header
     tags=["Facebook Action", "Messenger"],
     summary="Meta Messenger webhook: POST signed messaging events",
 )
@@ -644,7 +642,7 @@ async def messenger_interact_webhook_events(request: Request, agent_id: str) ->
     agent, fb_action = await _agent_and_facebook_action_for_messenger_webhook(agent_id)
 
     fb_action._apply_env_defaults()
-    app_secret = str(fb_action.app_secret or "").strip()
+    app_secret = str(fb_action._app_secret() or "").strip()
     if not app_secret:
         raise HTTPException(
             status_code=500, detail="FACEBOOK_APP_SECRET is required for webhook POST"
diff --git a/jvagent/action/facebook_action/facebook_action.py b/jvagent/action/facebook_action/facebook_action.py
index 5dc3ffd2..b2db9d43 100644
--- a/jvagent/action/facebook_action/facebook_action.py
+++ b/jvagent/action/facebook_action/facebook_action.py
@@ -29,6 +29,10 @@
 # ones). AUDIT-actions (LOW).
 _BACKGROUND_TASKS: set = set()
 
+# Track which action IDs have already scheduled a deferred messenger webhook
+# registration via the jvspatial lifecycle hook (prevents double-registration).
+_messenger_webhook_startup_hooks: set = set()
+
 
 class FacebookAction(Action):
     """Action for Facebook Graph API (page management, Messenger, webhooks)."""
@@ -666,9 +670,146 @@ async def on_startup(self) -> None:
                 "Use admin GET .../facebook/messenger/webhook-url or register manually."
             )
         elif self.base_url and str(self.base_url).strip():
+            self._schedule_deferred_messenger_webhook_register()
+
+    @staticmethod
+    def _startup_webhook_register_timeout_seconds() -> float:
+        """Fail-fast bound for startup Messenger webhook registration (seconds)."""
+        try:
+            raw = os.environ.get(
+                "FACEBOOK_STARTUP_WEBHOOK_REGISTER_TIMEOUT_SECONDS", ""
+            ).strip()
+            if raw:
+                return max(1.0, float(raw))
+        except (ValueError, TypeError):
+            pass
+        return 60.0
+
+    async def _run_startup_messenger_webhook_register(self) -> None:
+        """Health-check then register Messenger webhook (startup path only).
+
+        Unlike ``on_reload``, this path is reserved for first boot and will
+        always attempt registration if not already done.
+        """
+        try:
+            if not self.webhook_url or "?api_key=" not in (self.webhook_url or ""):
+                await self.get_webhook_url()
+
+            if not self.webhook_url:
+                logger.warning(
+                    "FacebookAction: cannot register webhook — webhook_url is empty"
+                )
+                return
+
+            logger.info("Registering Facebook Messenger webhook on startup")
+            reg = await self.register_messenger_webhook_subscription()
+            if reg.get("status") == "ok":
+                logger.info(
+                    "Facebook Messenger webhook registration succeeded: " "callback=%s",
+                    reg.get("callback_url") or self.webhook_url,
+                )
+            elif reg.get("status") == "skipped":
+                logger.info(
+                    "Facebook Messenger webhook registration skipped: %s",
+                    reg.get("reason"),
+                )
+            else:
+                logger.warning(
+                    "Facebook Messenger webhook registration: %s",
+                    reg,
+                )
+        except Exception as e:
+            logger.warning(
+                "Facebook Messenger startup webhook register failed: %s",
+                e,
+                exc_info=True,
+            )
+
+    def _schedule_deferred_messenger_webhook_register(self) -> None:
+        """Register Messenger webhook after the HTTP server is listening.
+
+        ``on_startup`` runs inside ``asyncio.run(pre_startup_bootstrap)``; a bare
+        ``asyncio.create_task`` there may fire before uvicorn is listening, causing
+        Meta's verification GET to fail. Hook into the jvspatial server lifecycle
+        instead (same pattern as WhatsAppAction). Falls back to asyncio.create_task
+        if lifecycle_manager is unavailable.
+        """
+        action_id = str(getattr(self, "id", "") or "")
+        if action_id and action_id in _messenger_webhook_startup_hooks:
+            logger.debug(
+                "FacebookAction id=%s: deferred webhook register already scheduled",
+                action_id,
+            )
+            return
+
+        try:
+            from jvspatial.api.context import get_current_server
+
+            server = get_current_server()
+            if not server or not hasattr(server, "lifecycle_manager"):
+                logger.warning(
+                    "FacebookAction: cannot schedule webhook registration via "
+                    "lifecycle_manager (server not ready); falling back to asyncio.create_task"
+                )
+                raise RuntimeError("lifecycle_manager unavailable")
 
             async def _deferred_messenger_webhook_register() -> None:
-                """Let the HTTP server and any tunnel come up before Meta probes the URL."""
+                """Schedule Messenger webhook registration after uvicorn startup."""
+                try:
+                    delay_raw = os.environ.get(
+                        "FACEBOOK_WEBHOOK_REGISTER_DELAY_SECONDS", "0"
+                    )
+                    delay_sec = max(0.0, float(delay_raw))
+                except (ValueError, TypeError):
+                    delay_sec = 0.0
+
+                async def _run_after_startup() -> None:
+                    if delay_sec > 0:
+                        logger.info(
+                            "Deferring Facebook Messenger webhook registration by %.1fs "
+                            "(after Application startup complete)",
+                            delay_sec,
+                        )
+                        await asyncio.sleep(delay_sec)
+                    else:
+                        await asyncio.sleep(0)
+
+                    timeout = self._startup_webhook_register_timeout_seconds()
+                    try:
+                        await asyncio.wait_for(
+                            self._run_startup_messenger_webhook_register(),
+                            timeout=timeout,
+                        )
+                    except asyncio.TimeoutError:
+                        logger.warning(
+                            "Facebook Messenger startup webhook register timed out "
+                            "after %.1fs. Set FACEBOOK_SKIP_STARTUP_WEBHOOK_REGISTRATION=true "
+                            "or register via admin POST .../facebook/webhook/register.",
+                            timeout,
+                        )
+
+                asyncio.create_task(
+                    _run_after_startup(),
+                    name="facebook_messenger_webhook_register",
+                )
+
+            server.lifecycle_manager.add_startup_hook(
+                _deferred_messenger_webhook_register
+            )
+            if action_id:
+                _messenger_webhook_startup_hooks.add(action_id)
+            logger.info(
+                "FacebookAction id=%s: scheduled deferred webhook registration "
+                "via lifecycle_manager",
+                action_id,
+            )
+        except Exception:
+            logger.warning(
+                "FacebookAction: failed to schedule deferred webhook registration "
+                "via lifecycle_manager; falling back to asyncio.create_task"
+            )
+
+            async def _fallback_deferred() -> None:
                 try:
                     delay_raw = os.environ.get(
                         "FACEBOOK_WEBHOOK_REGISTER_DELAY_SECONDS", "8"
@@ -678,28 +819,13 @@ async def _deferred_messenger_webhook_register() -> None:
                     delay_sec = 8.0
                 if delay_sec > 0:
                     logger.info(
-                        "Deferring Meta webhook subscription by %.1fs (%s)",
+                        "Deferring Meta webhook subscription by %.1fs (fallback)",
                         delay_sec,
-                        "FACEBOOK_WEBHOOK_REGISTER_DELAY_SECONDS",
                     )
                     await asyncio.sleep(delay_sec)
-                if not self.webhook_url or "?api_key=" not in (self.webhook_url or ""):
-                    logger.warning(
-                        "FacebookAction id=%s: webhook_url still empty before deferred "
-                        "Meta subscribe; register_messenger_webhook_subscription will try "
-                        "get_webhook_url",
-                        getattr(self, "id", None),
-                    )
-                reg = await self.register_messenger_webhook_subscription()
-                if reg.get("status") not in ("ok",):
-                    logger.warning(
-                        "Facebook deferred webhook registration: %s",
-                        reg,
-                    )
+                await self._run_startup_messenger_webhook_register()
 
-            # Retain a strong reference: asyncio holds only a weak one, so a bare
-            # create_task() can be GC'd mid-flight. AUDIT-actions (LOW).
-            _task = asyncio.create_task(_deferred_messenger_webhook_register())
+            _task = asyncio.create_task(_fallback_deferred())
             _BACKGROUND_TASKS.add(_task)
             _task.add_done_callback(_BACKGROUND_TASKS.discard)
 
diff --git a/jvagent/action/pageindex/endpoints.py b/jvagent/action/pageindex/endpoints.py
index c88b7ed4..0ded31ee 100644
--- a/jvagent/action/pageindex/endpoints.py
+++ b/jvagent/action/pageindex/endpoints.py
@@ -1738,12 +1738,14 @@ async def delete_document_chunk_endpoint(
                 description="Search results with content and document metadata",
                 example=[
                     {
-                        "node_id": "n.DocumentNode.xyz",
-                        "title": "Section Title",
                         "doc_name": "my_doc",
-                        "content": "Excerpt...",
+                        "title": "Section Title",
+                        "start_page": 5,
+                        "end_page": 8,
                         "start_index": 5,
                         "end_index": 8,
+                        "node_id": "n.DocumentNode.xyz",
+                        "content": "Excerpt...",
                         "doc_url": "https://example.com/doc.pdf",
                     }
                 ],
@@ -1796,7 +1798,12 @@ async def search_documents_endpoint(
     | only_enabled | bool | No | When true (default), skip disabled chunks |
     | include | string[] | No | Extra fields per hit (e.g. hierarchy, content_type, pageindex_node_id) |
 
-    **Response:** `results` — array of `{node_id, title, doc_name, content, text, summary, start_index, end_index, physical_index, doc_url}`
+    **Response:** `results` — array of `{doc_name, title, start_page, end_page, start_index, end_index, node_id, structure, content, summary, doc_url}`
+
+    Citation/identity fields come first so truncation keeps them. Bulk `text`
+    (and `physical_index` / `enabled`) are omitted by default — request them
+    per hit with `include=["text", "physical_index", "enabled"]`. `start_page` /
+    `end_page` are aliases of `start_index` / `end_index`, both returned.
 
     Collection is determined by `agent_id` from the path.
     """
diff --git a/jvagent/action/pageindex/models.py b/jvagent/action/pageindex/models.py
index 3e927ac1..ac145fe6 100644
--- a/jvagent/action/pageindex/models.py
+++ b/jvagent/action/pageindex/models.py
@@ -147,18 +147,29 @@ def node_to_result(
             or node.title
             or ""
         )
+    # Citation/identity fields first so orchestrator middle-elision keeps them;
+    # omit physical_index/enabled/text; cap large bodies last.
+    #
+    # ``start_index`` / ``end_index`` are retained alongside the friendlier
+    # ``start_page`` / ``end_page`` because these rows are returned verbatim by
+    # the public search endpoint (``endpoints.py`` -> ``{"results": results}``).
+    # They are two small ints, so keeping them costs nothing against the
+    # observation budget — the bulk savings come from omitting ``text``.
     return {
-        "node_id": node.id,
-        "title": node.title,
-        "text": node.text,
-        "summary": node.summary,
         "doc_name": node.doc_name,
-        "structure": node.structure,
-        "content": content[:_MAX_CONTENT_CHARS] if content else "",
+        "title": node.title,
+        "start_page": node.start_index,
+        "end_page": node.end_index,
         "start_index": node.start_index,
         "end_index": node.end_index,
-        "physical_index": node.physical_index,
-        "enabled": node_enabled(node),
+        "node_id": node.id,
+        "structure": node.structure,
+        "content": content[:_MAX_CONTENT_CHARS] if content else "",
+        "summary": (
+            (node.summary or "")[:_MAX_CONTENT_CHARS]
+            if node.summary is not None
+            else None
+        ),
     }
 
 
@@ -172,6 +183,8 @@ def node_to_result(
     "line_num": lambda n: n.line_num,
     "start_index": lambda n: n.start_index,
     "end_index": lambda n: n.end_index,
+    "start_page": lambda n: n.start_index,
+    "end_page": lambda n: n.end_index,
     "physical_index": lambda n: n.physical_index,
     "enabled": lambda n: node_enabled(n),
     "content_type": lambda n: getattr(n, "content_type", None),
diff --git a/jvagent/action/pageindex/pageindex_action/info.yaml b/jvagent/action/pageindex/pageindex_action/info.yaml
index ca587208..3546f61d 100644
--- a/jvagent/action/pageindex/pageindex_action/info.yaml
+++ b/jvagent/action/pageindex/pageindex_action/info.yaml
@@ -14,9 +14,6 @@ package:
     doc_description: false
     max_token_num_each_node: 20000
     summary_token_threshold: 200
-    limit: 10
-    strategy: tree_search
-    include_references: true
   dependencies:
     jvagent: ~0.0.1
     actions: []
diff --git a/jvagent/action/pageindex/pageindex_action/pageindex_action.py b/jvagent/action/pageindex/pageindex_action/pageindex_action.py
index dc24e143..dd500d0d 100644
--- a/jvagent/action/pageindex/pageindex_action/pageindex_action.py
+++ b/jvagent/action/pageindex/pageindex_action/pageindex_action.py
@@ -605,7 +605,10 @@ async def _t_search(
         doc_name: Annotated[
             Optional[str], "Restrict search to a specific document name."
         ] = None,
-        limit: Annotated[int, "Max results to return (default 5)."] = 5,
+        limit: Annotated[
+            Optional[int],
+            "Max results to return. Uses the action's configured limit when omitted.",
+        ] = None,
         **kwargs: Any,
     ) -> str:
         """Search the internal knowledge base for documents matching a query."""
@@ -764,12 +767,15 @@ async def _t_list_docs(
         ] = None,
         summary: Annotated[
             bool,
-            "If true, return only document names and descriptions (lighter response for quick lookup).",
-        ] = False,
+            "If true (default), return only document names and short descriptions "
+            "(lighter response for quick lookup). If false, also include access "
+            "(and chunks when the catalog still fits the observation budget).",
+        ] = True,
     ) -> str:
         """List documents in the knowledge base. When access control is enabled,
-        only documents the current user can access are returned. Set summary=true
-        to get document names and descriptions for quick lookup."""
+        only documents the current user can access are returned. Returns a compact
+        ``{count, documents}`` payload (truncated descriptions) so the full catalog
+        fits in the orchestrator observation budget."""
         import json
 
         from jvagent.tooling.tool_executor import get_tool_visitor
@@ -782,7 +788,72 @@ async def _t_list_docs(
             session_id=getattr(visitor, "session_id", None) if visitor else None,
             summary=summary,
         )
-        return json.dumps(result, indent=2)
+        # Lean tool payload: keep under the orchestrator's observation budget so
+        # it does not middle-elide catalog entries.
+        from jvagent.action.orchestrator.tools import DEFAULT_OBSERVATION_MAX_CHARS
+
+        max_chars = DEFAULT_OBSERVATION_MAX_CHARS
+        desc_limit = 80
+        total = len(result)
+
+        def _build(limit: int, include_chunks: bool, keep: Optional[int]) -> list:
+            docs: List[Dict[str, Any]] = []
+            for doc in result[:keep] if keep is not None else result:
+                desc = doc.get("doc_description", "") or ""
+                if len(desc) > limit:
+                    desc = desc[: limit - 3] + "..."
+                entry: Dict[str, Any] = {
+                    "doc_name": doc.get("doc_name", ""),
+                    "doc_description": desc,
+                }
+                if not summary:
+                    if include_chunks and "chunks" in doc:
+                        entry["chunks"] = doc["chunks"]
+                    meta = doc.get("metadata")
+                    if isinstance(meta, dict) and "access" in meta:
+                        entry["access"] = meta["access"]
+                docs.append(entry)
+            return docs
+
+        def _dump(docs: list) -> str:
+            body: Dict[str, Any] = {"count": total, "documents": docs}
+            if len(docs) < total:
+                # Say so explicitly: a model told it has 60 of 200 can ask for
+                # the rest, one silently handed 60 cannot.
+                body["shown"] = len(docs)
+                body["truncated"] = True
+            return json.dumps(body, separators=(",", ":"))
+
+        include_chunks = not summary
+        keep: Optional[int] = None
+        documents = _build(desc_limit, include_chunks, keep)
+        payload = _dump(documents)
+        # Prefer dropping chunks over cutting names; then shrink descriptions.
+        if len(payload) > max_chars and include_chunks:
+            include_chunks = False
+            documents = _build(desc_limit, include_chunks, keep)
+            payload = _dump(documents)
+        while len(payload) > max_chars and desc_limit > 40:
+            desc_limit -= 10
+            documents = _build(desc_limit, include_chunks, keep)
+            payload = _dump(documents)
+        # Descriptions are as short as they go and it still does not fit, so the
+        # catalog itself is too long. Drop whole entries rather than return an
+        # oversized payload for the orchestrator to elide from the middle.
+        while len(payload) > max_chars and len(documents) > 1:
+            keep = max(1, len(documents) - max(1, len(documents) // 10))
+            documents = _build(desc_limit, include_chunks, keep)
+            payload = _dump(documents)
+        logger.info(
+            "pageindex__list returned %d of %d document(s) (summary=%s, "
+            "collection=%s, payload_chars=%d)",
+            len(documents),
+            total,
+            summary,
+            collection_name or self._resolve_collection(),
+            len(payload),
+        )
+        return payload
 
     @tool(name="pageindex__delete")
     async def _t_delete_doc(
diff --git a/jvagent/action/pageindex/pageindex_action/runtime_config.py b/jvagent/action/pageindex/pageindex_action/runtime_config.py
index 976a1555..6d7bfd95 100644
--- a/jvagent/action/pageindex/pageindex_action/runtime_config.py
+++ b/jvagent/action/pageindex/pageindex_action/runtime_config.py
@@ -92,8 +92,12 @@ def normalize_retrieval_excerpt_source(value: Any, fallback: str) -> str:
 
 def format_page_range(r: Dict[str, Any]) -> str:
     """Format page range from result dict, e.g. 'pp. 5-8' or 'p. 5'."""
-    start = r.get("start_index")
-    end = r.get("end_index")
+    # ``dict.get(key, default)`` only falls back when the key is absent, and a
+    # row may carry an explicit ``None``; fall back on falsy-None too.
+    start = r.get("start_page")
+    start = r.get("start_index") if start is None else start
+    end = r.get("end_page")
+    end = r.get("end_index") if end is None else end
     if start is not None and end is not None and start != end:
         return f"pp. {start}-{end}"
     if start is not None:
diff --git a/jvagent/action/parameters.py b/jvagent/action/parameters.py
index 91d5fc20..6a367ca7 100644
--- a/jvagent/action/parameters.py
+++ b/jvagent/action/parameters.py
@@ -737,7 +737,6 @@ def render_parameters(parameters: Optional[List[Any]]) -> str:
         r"anything|need anything|further)\b",
         re.I,
     ),
-    re.compile(r"\blet me know if\b", re.I),
     re.compile(r"\bhope (this|that|it)\b[^.!?]*\bhelps?\b", re.I),
 ]
 
@@ -894,10 +893,26 @@ def _detect_drop_internal_disclosure(text: str) -> str:
     return _drop_matching(text, _INTERNAL_NAME_PATTERNS)
 
 
+def _is_question(sentence: str) -> bool:
+    """True if the sentence ends in a question mark.
+
+    Questions are never closers — stripping "Could you let me know if you
+    need a quote?" is a false positive. Trailing quotes, brackets, and
+    emphasis are ignored so 'Need a hand?"' and 'Shall we?)' still count.
+    """
+    s = sentence.strip().rstrip("\"'’”)]}*_ \t")
+    return s.endswith("?") or s.endswith("?!")
+
+
 def _detect_peel_closers(text: str) -> str:
     """Scrub detector for ``voice.closers``."""
     kept = [m.group(0) for m in _SENTENCE_RE.finditer(text)]
-    while len(kept) > 1 and kept[-1].strip() and _is_closer(kept[-1]):
+    while (
+        len(kept) > 1
+        and kept[-1].strip()
+        and _is_closer(kept[-1])
+        and not _is_question(kept[-1])
+    ):
         kept.pop()
     return "".join(kept)
 
@@ -961,7 +976,20 @@ def vet_egress(
             continue
         for detector in detectors_for(param, ENFORCEMENT_SCRUB):
             try:
+                before = cleaned
                 cleaned = detector(cleaned)
+                if cleaned != before:
+                    logger.debug(
+                        "parameters: scrub detector %r for %r shortened text (%d -> %d chars)",
+                        (
+                            detector.__name__
+                            if hasattr(detector, "__name__")
+                            else detector
+                        ),
+                        param.get("key"),
+                        len(before),
+                        len(cleaned),
+                    )
             except Exception as exc:  # pragma: no cover - defensive
                 logger.warning(
                     "parameters: scrub detector for %r failed: %s",
diff --git a/tests/action/artifact_handler_interact_action/test_publish_messenger_message.py b/tests/action/artifact_handler_interact_action/test_publish_messenger_message.py
new file mode 100644
index 00000000..7585d5bc
--- /dev/null
+++ b/tests/action/artifact_handler_interact_action/test_publish_messenger_message.py
@@ -0,0 +1,116 @@
+"""_publish_messenger_message uses the registered MessengerAdapter's live action."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from jvagent.action.artifact_handler_interact_action import endpoints as ep
+
+
+class _FakeInteraction:
+    def add_parameter(self, *_a, **_k):
+        return None
+
+    def set_response(self, *_a, **_k):
+        return None
+
+    async def save(self):
+        return self
+
+
+class _FakeConversation:
+    session_id = "sess-1"
+
+    async def add_interaction(self, **_kwargs):
+        return _FakeInteraction()
+
+
+@pytest.mark.asyncio
+async def test_publish_messenger_uses_registered_adapter_action(monkeypatch):
+    """Happy path: use adapter.action.api().send_text_message, not find_one."""
+    send_text = MagicMock(return_value={"message_id": "m1"})
+    live_action = SimpleNamespace(
+        is_configured=MagicMock(return_value=True),
+        api=MagicMock(return_value=SimpleNamespace(send_text_message=send_text)),
+    )
+    adapter = SimpleNamespace(_initialized=True, action=live_action)
+    response_bus = SimpleNamespace(_channel_adapters={"messenger": adapter})
+
+    agent = SimpleNamespace(
+        get_memory=AsyncMock(
+            return_value=SimpleNamespace(
+                get_user=AsyncMock(return_value=None),
+            )
+        ),
+        get_response_bus=AsyncMock(return_value=response_bus),
+        get_action_by_type=AsyncMock(side_effect=AssertionError("must not find_one")),
+    )
+
+    monkeypatch.setattr(
+        "jvagent.memory.conversation.Conversation.get",
+        AsyncMock(return_value=_FakeConversation()),
+    )
+
+    ok = await ep._publish_messenger_message(
+        agent=agent,
+        user_id="psid-1",
+        session_id="sess-1",
+        conversation_id="conv-1",
+        content="Document ready.",
+        display_doc="doc.pdf",
+        job_id="job-1",
+        answered=False,
+    )
+    assert ok is True
+    send_text.assert_called_once_with("psid-1", "Document ready.")
+    agent.get_action_by_type.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_publish_messenger_cold_start_registers_adapter(monkeypatch):
+    """When adapter missing, resolve token, register, then send via live action."""
+    send_text = MagicMock(return_value={"message_id": "m2"})
+    live_action = SimpleNamespace(
+        is_configured=MagicMock(return_value=True),
+        api=MagicMock(return_value=SimpleNamespace(send_text_message=send_text)),
+        ensure_page_access_token=AsyncMock(return_value={"updated": True}),
+        ensure_adapter_registered=AsyncMock(return_value=True),
+    )
+    adapters: dict = {}
+    response_bus = SimpleNamespace(_channel_adapters=adapters)
+
+    async def _register():
+        adapters["messenger"] = SimpleNamespace(_initialized=True, action=live_action)
+        return True
+
+    live_action.ensure_adapter_registered = AsyncMock(side_effect=_register)
+
+    agent = SimpleNamespace(
+        get_memory=AsyncMock(
+            return_value=SimpleNamespace(get_user=AsyncMock(return_value=None))
+        ),
+        get_response_bus=AsyncMock(return_value=response_bus),
+        get_action_by_type=AsyncMock(return_value=live_action),
+    )
+
+    monkeypatch.setattr(
+        "jvagent.memory.conversation.Conversation.get",
+        AsyncMock(return_value=_FakeConversation()),
+    )
+
+    ok = await ep._publish_messenger_message(
+        agent=agent,
+        user_id="psid-2",
+        session_id="sess-1",
+        conversation_id="conv-1",
+        content="Ready.",
+        display_doc="x.png",
+        job_id="job-2",
+    )
+    assert ok is True
+    live_action.ensure_page_access_token.assert_awaited_once()
+    live_action.ensure_adapter_registered.assert_awaited_once()
+    send_text.assert_called_once_with("psid-2", "Ready.")
diff --git a/tests/action/pageindex/test_pageindex.py b/tests/action/pageindex/test_pageindex.py
index 879de126..7b43db17 100644
--- a/tests/action/pageindex/test_pageindex.py
+++ b/tests/action/pageindex/test_pageindex.py
@@ -449,11 +449,25 @@ async def test_adapter_persists_distinct_summary_vs_text(pageindex_temp_db):
     assert len(results) >= 1
     node = next((r for r in results if r.get("doc_name") == "summary_test"), None)
     assert node is not None
-    assert node.get("text") == "Full section text content here"
+    assert "text" not in node
     assert node.get("summary") == "LLM-generated summary"
-    assert node["summary"] != node["text"]
     assert node.get("content") == "LLM-generated summary"
 
+    results_with_text = await search_documents(
+        query="section",
+        strategy="direct",
+        limit=5,
+        collection_name="col_summary",
+        include=["text"],
+    )
+    node_full = next(
+        (r for r in results_with_text if r.get("doc_name") == "summary_test"), None
+    )
+    assert node_full is not None
+    assert node_full.get("text") == "Full section text content here"
+    assert node_full.get("summary") == "LLM-generated summary"
+    assert node_full["summary"] != node_full["text"]
+
     set_pageindex_retrieval_excerpt_source("text")
     results_text = await search_documents(
         query="section",
@@ -479,6 +493,67 @@ def test_node_to_result_excerpt_source_explicit():
     assert node_to_result(n, excerpt_source="text")["content"] == "full body"
 
 
+def test_node_to_result_citation_fields_first_and_bodies_capped():
+    """Search hits put citation keys first; omit text/physical_index/enabled; cap bodies."""
+    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.start_index = 3
+    n.end_index = 5
+    n.structure = "1.2"
+    n.physical_index = 3
+    result = node_to_result(n, excerpt_source="summary")
+    keys = list(result)
+    assert keys.index("doc_name") < keys.index("content")
+    assert keys.index("doc_name") < keys.index("summary")
+    assert "text" not in result
+    assert "physical_index" not in result
+    assert "enabled" not in result
+    assert result["doc_name"] == "manual.pdf"
+    assert result["start_page"] == 3
+    assert result["end_page"] == 5
+    assert len(result["content"]) <= _MAX_CONTENT_CHARS
+    assert len(result["summary"]) == _MAX_CONTENT_CHARS
+    assert result["summary"] == "S" * _MAX_CONTENT_CHARS
+
+
+def test_node_to_result_keeps_index_keys_for_api_compatibility():
+    """These rows are returned verbatim by the search endpoint.
+
+    Renaming start_index -> start_page would break every existing API client
+    for no budget saving, so both spellings ship.
+    """
+    n = DocumentNode()
+    n.doc_name = "manual.pdf"
+    n.start_index = 3
+    n.end_index = 5
+    result = node_to_result(n)
+    assert result["start_index"] == 3
+    assert result["end_index"] == 5
+    assert result["start_page"] == result["start_index"]
+    assert result["end_page"] == result["end_index"]
+
+
+def test_format_page_range_falls_back_on_explicit_none():
+    """A row may carry start_page=None; dict.get's default would not fire."""
+    from jvagent.action.pageindex.pageindex_action.runtime_config import (
+        format_page_range,
+    )
+
+    assert format_page_range({"start_page": None, "start_index": 4}) == "p. 4"
+    assert (
+        format_page_range(
+            {"start_page": None, "end_page": None, "start_index": 4, "end_index": 6}
+        )
+        == "pp. 4-6"
+    )
+    assert format_page_range({"start_page": 7, "end_page": 9}) == "pp. 7-9"
+
+
 def test_parse_llm_json_object_ignores_trailing_text():
     raw = '{"thinking":"x","node_list":["0001"]}\n\nExtra thanks.'
     out = _parse_llm_json_object(raw)
@@ -2231,3 +2306,125 @@ async def test_search_access_public_plus_private_includes_both(
         "doc_public_tagged",
         "doc_private",
     }
+
+
+@pytest.mark.asyncio
+async def test_t_list_docs_compact_payload_fits_observation_budget(caplog):
+    """pageindex__list tool returns lean JSON so ~21 long descs fit under 4000 chars."""
+    import json
+    import logging
+
+    long_desc = (
+        "Quality Management System manual for Silvie's Industrial Solutions "
+        "covering departmental procedures, compliance, KPIs, and staff roles "
+        "across procurement, sales, accounts, and warehouse operations."
+    )
+    assert len(long_desc) > 80
+
+    listed = [
+        {
+            "doc_name": f"DOC_{i:02d} - SIS-MNL-{i:03d}.pdf",
+            "doc_description": long_desc,
+            "doc_url": f"https://docs.google.com/document/d/example{i}/edit",
+            "root_id": f"n.DocumentRootNode.{i:024d}",
+            "collection_name": "n.Agent.silvie",
+            "metadata": {"access": "private" if i % 2 else "public"},
+            "chunks": 100 + i,
+        }
+        for i in range(21)
+    ]
+
+    action = _make_pageindex_action(access_control=True)
+    object.__setattr__(action, "collection", "n.Agent.silvie")
+
+    with (
+        patch.object(
+            PageIndexAction,
+            "list_documents",
+            new_callable=AsyncMock,
+            return_value=listed,
+        ),
+        patch.object(
+            PageIndexAction,
+            "_resolve_collection",
+            return_value="n.Agent.silvie",
+        ),
+        patch(
+            "jvagent.tooling.tool_executor.get_tool_visitor",
+            return_value=None,
+        ),
+        caplog.at_level(
+            logging.INFO,
+            logger="jvagent.action.pageindex.pageindex_action.pageindex_action",
+        ),
+    ):
+        summary_payload = await PageIndexAction._t_list_docs(action, summary=True)
+        full_payload = await PageIndexAction._t_list_docs(action, summary=False)
+
+    for payload, expect_extra in ((summary_payload, False), (full_payload, True)):
+        assert len(payload) <= 4000, len(payload)
+        parsed = json.loads(payload)
+        assert parsed["count"] == 21
+        assert len(parsed["documents"]) == 21
+        names = [d["doc_name"] for d in parsed["documents"]]
+        assert names == [f"DOC_{i:02d} - SIS-MNL-{i:03d}.pdf" for i in range(21)]
+        for entry in parsed["documents"]:
+            assert len(entry["doc_description"]) <= 80
+            assert "doc_url" not in entry
+            assert "root_id" not in entry
+            assert "collection_name" not in entry
+            if expect_extra:
+                assert "access" in entry
+            else:
+                assert "chunks" not in entry
+                assert "access" not in entry
+
+    assert any(
+        "pageindex__list returned 21 of 21 document(s)" in r.message
+        and "payload_chars=" in r.message
+        for r in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_t_list_docs_drops_entries_rather_than_overflow_budget():
+    """A catalog too large to fit even at minimum description length.
+
+    Shrinking descriptions bottoms out at 40 chars; past that the only way to
+    stay under the observation budget is to return fewer entries. Doing that
+    silently would be worse than not truncating at all, so the payload has to
+    say what it dropped.
+    """
+    import json
+
+    from jvagent.action.orchestrator.tools import DEFAULT_OBSERVATION_MAX_CHARS
+
+    listed = [
+        {
+            "doc_name": f"LONG_DOCUMENT_NAME_NUMBER_{i:04d}_procedures_manual.pdf",
+            "doc_description": "D" * 200,
+        }
+        for i in range(400)
+    ]
+
+    action = _make_pageindex_action(access_control=False)
+
+    with (
+        patch.object(
+            PageIndexAction,
+            "list_documents",
+            new_callable=AsyncMock,
+            return_value=listed,
+        ),
+        patch.object(PageIndexAction, "_resolve_collection", return_value="col"),
+        patch("jvagent.tooling.tool_executor.get_tool_visitor", return_value=None),
+    ):
+        payload = await PageIndexAction._t_list_docs(action, summary=True)
+
+    assert len(payload) <= DEFAULT_OBSERVATION_MAX_CHARS
+    parsed = json.loads(payload)
+    # The true total survives even though the list does not.
+    assert parsed["count"] == 400
+    assert parsed["truncated"] is True
+    assert parsed["shown"] == len(parsed["documents"]) < 400
+    assert parsed["documents"][0]["doc_name"] == listed[0]["doc_name"]
diff --git a/tests/action/test_parameters.py b/tests/action/test_parameters.py
index 4a47a1d4..f7ab7c9b 100644
--- a/tests/action/test_parameters.py
+++ b/tests/action/test_parameters.py
@@ -167,6 +167,23 @@ def test_vet_egress_keeps_specific_ask_and_questions():
     assert vet_egress("Happy to help!") == "Happy to help!"
 
 
+def test_vet_egress_keeps_trailing_questions_that_look_like_closers():
+    """A closer-shaped sentence is still a question if it asks something.
+
+    Stripping it leaves the user with no prompt to answer, which is how the
+    agent ends up talking past them.
+    """
+    a = "Your quote is ready. Could you let me know if you need a quote?"
+    assert vet_egress(a) == a
+    # Trailing quotes/brackets must not hide the question mark.
+    b = 'Shipping is free. Want me to check stock?"'
+    assert vet_egress(b) == b
+    c = "That's covered. Anything else I can help with?)"
+    assert vet_egress(c) == c
+    d = "All set. Need anything further?!"
+    assert vet_egress(d) == d
+
+
 def test_vet_egress_preserves_newlines_between_list_items():
     # Markdown list items live on their own lines. The scrub must NOT weld
     # consecutive sentences into one run (regression: "city center.Jan Thiel").