fix(agent): isolate response language from connector context (toby)

src/xagent/core/agent/context/enrichment.py

@@ -2,7 +2,8 @@
 
 import asyncio
 import logging
-from typing import Any, cast
+from dataclasses import dataclass
+from typing import Any, Literal, cast
 
 from ...agent.trace import (
     trace_memory_retrieve_end,
@@ -21,6 +22,54 @@
 IMAGE_EDIT_UNAVAILABLE_METADATA_KEY = "image_edit_unavailable"
 
 
+DisplayMessageState = Literal["missing", "empty", "text"]
+TOP_LEVEL_USER_REQUEST_METADATA_KEY = "_xagent_top_level_user_request"
+
+
+@dataclass(frozen=True)
+class TopLevelUserRequest:
+    """One executable request and its presentation-only language boundary."""
+
+    execution_text: str
+    language_text: str
+    display_state: DisplayMessageState
+    has_pending_response: bool = False
+
+
+def _stored_top_level_user_request(context: Any) -> TopLevelUserRequest | None:
+    metadata = getattr(context, "metadata", None)
+    if not isinstance(metadata, dict):
+        return None
+    payload = metadata.get(TOP_LEVEL_USER_REQUEST_METADATA_KEY)
+    if not isinstance(payload, dict):
+        return None
+    execution_text = payload.get("execution_text")
+    language_text = payload.get("language_text")
+    display_state = payload.get("display_state")
+    if (
+        not isinstance(execution_text, str)
+        or not isinstance(language_text, str)
+        or display_state not in {"missing", "empty", "text"}
+    ):
+        return None
+    return TopLevelUserRequest(
+        execution_text=execution_text,
+        language_text=language_text,
+        display_state=display_state,
+    )
+
+
+def _persist_top_level_user_request(context: Any, request: TopLevelUserRequest) -> None:
+    metadata = getattr(context, "metadata", None)
+    if not isinstance(metadata, dict):
+        return
+    metadata[TOP_LEVEL_USER_REQUEST_METADATA_KEY] = {
+        "execution_text": request.execution_text,
+        "language_text": request.language_text,
+        "display_state": request.display_state,
+    }
+
+
 async def enrich_context_with_memory(
     *,
     context: Any,
@@ -110,23 +159,122 @@ def build_skill_context(skill: dict[str, Any]) -> str:
     return f"## Available Skill: {name}\n\n{content}".strip()
 
 
+def display_message_override(metadata: Any) -> str | None:
+    """Return a supported display-message override, including an empty one.
+
+    Missing keys and non-string values in directly constructed or restored
+    contexts keep the execution-content fallback. The production runner
+    normalizes a present non-string value to an authoritative empty string before
+    this helper. Any present string is authoritative after trimming, so file-only
+    turns do not expose augmented connector or attachment text as language evidence.
+    """
+    if not isinstance(metadata, dict) or "display_message" not in metadata:
+        return None
+    display = metadata["display_message"]
+    if not isinstance(display, str):
+        return None
+    return display.strip()
+
+
+def top_level_user_request(context: Any) -> TopLevelUserRequest:
+    """Return the latest independent request, excluding DAG and wait scaffolding.
+
+    A present display string is authoritative for language even when empty.
+    Answers to pending agent questions remain conversational context, but they do
+    not replace the independent request. Prompt policy may still honor an explicit
+    language-change instruction in such an answer.
+    """
+    has_pending_response = False
+    for message in reversed(getattr(context, "messages", []) or []):
+        if getattr(message, "role", None) != "user" or getattr(
+            message, "hidden", False
+        ):
+            continue
+        metadata = getattr(message, "metadata", None)
+        metadata = metadata if isinstance(metadata, dict) else {}
+        if metadata.get("response_to_waiting_for_user"):
+            has_pending_response = True
+            continue
+        if metadata.get("dag_step_id"):
+            continue
+
+        execution_text = str(getattr(message, "content", "") or "").strip()
+        display_text = display_message_override(metadata)
+        if display_text is None:
+            if not execution_text:
+                continue
+            request = TopLevelUserRequest(
+                execution_text=execution_text,
+                language_text=execution_text,
+                display_state="missing",
+                has_pending_response=has_pending_response,
+            )
+            _persist_top_level_user_request(context, request)
+            return request
+        request = TopLevelUserRequest(
+            execution_text=execution_text,
+            language_text=display_text,
+            display_state="text" if display_text else "empty",
+            has_pending_response=has_pending_response,
+        )
+        _persist_top_level_user_request(context, request)
+        return request
+
+    stored = _stored_top_level_user_request(context)
+    if stored is not None:
+        return TopLevelUserRequest(
+            execution_text=stored.execution_text,
+            language_text=stored.language_text,
+            display_state=stored.display_state,
+            has_pending_response=has_pending_response,
+        )
+
+    task = (
+        context.metadata.get("task")
+        if isinstance(getattr(context, "metadata", None), dict)
+        else None
+    )
+    task_text = str(task or "").strip()
+    request = TopLevelUserRequest(
+        execution_text=task_text,
+        language_text=task_text,
+        display_state="missing",
+        has_pending_response=has_pending_response,
+    )
+    _persist_top_level_user_request(context, request)
+    return request
+
+
+def language_prompt_message(message: Any) -> dict[str, Any]:
+    """Serialize one prompt payload message without duplicating its content."""
+    payload = {
+        "role": getattr(message, "role", None),
+        "content": getattr(message, "content", None),
+    }
+    metadata = getattr(message, "metadata", None)
+    if (
+        payload["role"] == "user"
+        and isinstance(metadata, dict)
+        and metadata.get("response_to_waiting_for_user")
+    ):
+        payload["user_message_context"] = "pending_agent_question_response"
+    return payload
+
+
 def latest_user_text(context: Any, *, prefer_display: bool = False) -> str:
     """Return the latest user turn's text.
 
-    ``prefer_display`` returns what the user actually typed instead of the
-    runtime-augmented execution prompt; language anchors must use it, work
-    anchors must not.
+    ``prefer_display`` returns a present string ``display_message`` (including
+    an intentionally empty one) instead of the runtime-augmented execution
+    prompt. Missing values, plus non-string values in direct/restored contexts,
+    fall back to content. Language anchors must prefer display text; work anchors
+    must not.
     """
     for message in reversed(getattr(context, "messages", []) or []):
         if getattr(message, "role", None) == "user":
             if prefer_display:
-                metadata = getattr(message, "metadata", None)
-                display = (
-                    metadata.get("display_message")
-                    if isinstance(metadata, dict)
-                    else None
-                )
-                if isinstance(display, str) and display.strip():
+                display = display_message_override(getattr(message, "metadata", None))
+                if display is not None:
                     return display
             return str(getattr(message, "content", "") or "")
     task = context.metadata.get("task") if hasattr(context, "metadata") else None

src/xagent/core/agent/context/execution.py

@@ -42,6 +42,7 @@
     IMAGE_EDIT_UNAVAILABLE_METADATA_KEY,
     MEMORY_CONTEXT_METADATA_KEY,
     SKILL_CONTEXT_METADATA_KEY,
+    top_level_user_request,
 )
 from .memory_tool import MEMORY_TOOLS_METADATA_KEY
 from .message import LLMCallRecord, Message
@@ -518,41 +519,34 @@ def _current_time_context(self) -> str:
     def _current_user_request_text(self, *, prefer_display: bool = False) -> str:
         """Return the current request text.
 
-        ``prefer_display`` yields the user-typed message instead of the
-        execution prompt, whose appended file-reference block is fixed English
-        and would otherwise decide the language of a short foreign request.
+        ``prefer_display`` yields a present string ``display_message``, including
+        an intentionally empty one, instead of the execution prompt. Missing and
+        legacy non-string values fall back to content. This keeps appended file
+        or connector context from deciding the response language.
         """
-        for message in reversed(self.messages):
-            if message.hidden or message.role != "user":
-                continue
-            if message.metadata.get("response_to_waiting_for_user"):
-                continue
-            # A DAG child context copies the root messages and then appends step
-            # scaffolding; only the root request may anchor the response language.
-            if message.metadata.get("dag_step_id"):
-                continue
-            if prefer_display:
-                display = str(message.metadata.get("display_message") or "").strip()
-                if display:
-                    return display
-            content = str(message.content or "").strip()
-            if content:
-                return content
-        return str(self.metadata.get("task") or "").strip()
+        request = top_level_user_request(self)
+        return request.language_text if prefer_display else request.execution_text
 
     def _system_context(self) -> str:
         parts = [self._current_time_context(), FILE_REF_MODEL_INSTRUCTIONS]
         dag_step_id = self.metadata.get("dag_step_id")
-        current_task = self._current_user_request_text()
+        request = top_level_user_request(self)
+        current_task = request.execution_text
         output_language = effective_output_language(self)
-        if current_task and not dag_step_id:
+        if not dag_step_id and (current_task or request.display_state != "missing"):
+            language_request = request.language_text
             language_directives = output_language_directives(
-                output_language, section="root_system_context"
+                output_language,
+                section=(
+                    "root_existing_request"
+                    if request.display_state == "missing"
+                    else "root_system_context"
+                ),
+                request=language_request,
             )
             parts.append(
-                "Current user request:\n"
-                f"{current_task}\n\n"
-                "Conversation focus rules: answer the current user request above. "
+                "Conversation focus rules: answer the latest independent user "
+                "request in the conversation. "
                 "Earlier user and assistant messages are context only; use them to "
                 "resolve references and preserve continuity, but do not re-answer "
                 "previous requests or repeat previous final answers unless the "
@@ -625,7 +619,11 @@ def _system_context(self) -> str:
             request_anchor = output_language_directives(
                 output_language,
                 section="dag_step_request_anchor",
-                request=self._current_user_request_text(prefer_display=True),
+                request=(
+                    request.language_text
+                    if request.display_state != "missing"
+                    else None
+                ),
             )
             if request_anchor:
                 parts.append(request_anchor)
@@ -909,6 +907,9 @@ def create_child_context(
         include_system_prompt: bool = True,
         metadata: dict[str, Any] | None = None,
     ) -> "ExecutionContext":
+        # Child compaction may discard the copied root message, so snapshot its
+        # clean request provenance before metadata is cloned.
+        top_level_user_request(self)
         child_metadata = dict(self.metadata)
         if metadata:
             child_metadata.update(metadata)
@@ -1078,6 +1079,7 @@ def compact_if_needed(self, llm: Any = None) -> CompactResult:
                 strategy="none",
             )
 
+        top_level_user_request(self)
         total_tokens = self._get_total_tokens()
         if total_tokens > self.compact_config.threshold:
             result = self._compact(llm)
@@ -1094,6 +1096,7 @@ def build_llm_compact_request_if_needed(self) -> dict[str, Any] | None:
         if not self.compact_config.enabled:
             return None
 
+        top_level_user_request(self)
         total_tokens = self._get_total_tokens()
         if total_tokens <= self.compact_config.threshold:
             return None
@@ -1170,6 +1173,7 @@ def compact_with_llm_response(
         llm: Any = None,
         original_tokens: int | None = None,
     ) -> CompactResult:
+        top_level_user_request(self)
         original_count = len(self.messages)
         summary = (
             ""
@@ -1516,6 +1520,17 @@ def estimate_context_tokens(self) -> int:
         return self._get_total_tokens()
 
     def _get_total_tokens(self) -> int:
+        rendered_estimate = self._estimate_message_tokens(self.messages) + max(
+            1,
+            len(
+                "\n\n".join(
+                    part
+                    for part in (self.system_prompt, self._system_context())
+                    if part
+                )
+            )
+            // 4,
+        )
         if self.llm_calls:
             latest_call = self.llm_calls[-1]
             if latest_call.input_tokens > 0:
@@ -1533,8 +1548,11 @@ def _get_total_tokens(self) -> int:
                     delta_chars = self._message_content_chars(
                         self.messages[prompt_message_count:]
                     )
-                    return latest_call.input_tokens + max(0, delta_chars // 4)
-        return self._estimate_message_tokens(self.messages)
+                    return max(
+                        rendered_estimate,
+                        latest_call.input_tokens + max(0, delta_chars // 4),
+                    )
+        return rendered_estimate
 
     def _estimate_message_tokens(self, messages: list[Message]) -> int:
         return sum(