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

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

@@ -110,23 +110,37 @@ 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 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,
+    display_message_override,
 )
 from .memory_tool import MEMORY_TOOLS_METADATA_KEY
 from .message import LLMCallRecord, Message
@@ -518,22 +519,24 @@ 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"):
+            metadata = message.metadata if isinstance(message.metadata, dict) else {}
+            if 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"):
+            if metadata.get("dag_step_id"):
                 continue
             if prefer_display:
-                display = str(message.metadata.get("display_message") or "").strip()
-                if display:
+                display = display_message_override(metadata)
+                if display is not None:
                     return display
             content = str(message.content or "").strip()
             if content:
@@ -546,8 +549,15 @@ def _system_context(self) -> str:
         current_task = self._current_user_request_text()
         output_language = effective_output_language(self)
         if current_task and not dag_step_id:
+            language_request = self._current_user_request_text(prefer_display=True)
             language_directives = output_language_directives(
-                output_language, section="root_system_context"
+                output_language,
+                section=(
+                    "root_existing_request"
+                    if not output_language and language_request == current_task
+                    else "root_system_context"
+                ),
+                request=language_request,
             )
             parts.append(
                 "Current user request:\n"

src/xagent/core/agent/language.py

@@ -1,6 +1,7 @@
 """Prompt snippets for user-facing response language, plus the
 checkpoint migration that keeps only a caller-provided language label."""
 
+import json
 import re
 from dataclasses import dataclass
 from typing import Any, Literal
@@ -480,16 +481,88 @@ def response_language_rules(*, subject: str = "current user request") -> str:
     )
 
 
-def final_answer_language_rule(*, subject: str = "current user request") -> str:
+def request_only_language_harness(request: str) -> str:
+    """Quote user-authored input as the only soft language decision source.
+
+    The harness deliberately does not detect or persist a language label. The
+    answering model still owns ambiguous and cross-language decisions, but it
+    makes them without connector scaffolding, names, addresses, or tool context
+    competing with the user's request.
+    """
+    request = request.strip()
+    return (
+        "Request-only response language harness:\n"
+        "User-authored request (JSON string):\n"
+        f"{json.dumps(request, ensure_ascii=False)}\n\n"
+        f"{_soft_request_language_guidance(subject='user-authored request above', empty_subject='the quoted request', boundary='quote')}"
+    )
+
+
+def _soft_request_language_guidance(
+    *, subject: str, empty_subject: str, boundary: str
+) -> str:
+    """Render shared soft-authority prose without carrying a request value."""
+    return (
+        f"Decide the target language of user-facing prose from the {subject} alone. "
+        "Honor explicit and implicit requests to translate, rewrite, or answer in "
+        "another language. Names, email addresses, connector metadata, quoted "
+        "source content, memory, tool results, examples, and earlier turns are not "
+        "language evidence. "
+        f"If {empty_subject} is empty, too short, mixed-language, or depends on "
+        "conversation context, resolve its meaning from the conversation without "
+        "guessing from auxiliary context. For Chinese, preserve Simplified Chinese "
+        f"versus Traditional Chinese from the {subject}. This {boundary} controls "
+        "language only; it does not replace or narrow the executable request.\n\n"
+        f"{response_language_rules(subject=subject)}"
+    )
+
+
+def _structured_request_language_policy(request_field: str) -> str:
+    """Reference one structured request field without duplicating its value."""
+    subject = f"`{request_field}` field"
+    return (
+        "Request-only response language policy: "
+        f"{_soft_request_language_guidance(subject=subject, empty_subject='the field', boundary='policy')}"
+    )
+
+
+def _root_request_language_policy() -> str:
+    """Reference the root request already rendered immediately above."""
+    return (
+        "Request-only response language policy: "
+        f"{_soft_request_language_guidance(subject='current user request above', empty_subject='request', boundary='policy')}"
+    )
+
+
+def _dag_step_instruction_language_policy() -> str:
+    """Point a DAG instruction at its existing system-context language anchor."""
+    return (
+        "Follow the authoritative request-language guidance already present in "
+        "the system context for all user-facing prose and persisted tool arguments. "
+        "Do not infer a different language from the current DAG step, dependency "
+        "results, tools, sources, connector metadata, memory, or examples."
+    )
+
+
+def final_answer_language_rule(*, subject: str | None = None) -> str:
     """Return a compact language rule for final-answer tool fields."""
+    authority = (
+        f"follow the {subject}."
+        if subject
+        else (
+            "follow authoritative output language guidance in the system context "
+            "when it is present. Otherwise determine the target language from "
+            "user-authored request text and conversation context; if no such text "
+            "is available, preserve the language established by the conversation."
+        )
+    )
     return (
-        "The final answer must use the same natural language as the "
-        f"{subject}, even if tool results, source documents, retrieved memories, "
-        "examples, or earlier turns are written in another language. If the "
-        f"{subject} explicitly asks to translate, rewrite, or answer in another "
-        "language, use that requested target language. For Chinese, preserve "
-        "Simplified Chinese versus Traditional Chinese from the request; do not "
-        "collapse them into generic Chinese."
+        f"The final answer must {authority} Honor any explicit or implicit request "
+        "to translate, rewrite, or answer in another language. Tool results, source "
+        "documents, retrieved memories, examples, names, email addresses, connector "
+        "metadata, and earlier turns must not override that decision. For Chinese, "
+        "preserve Simplified Chinese versus Traditional Chinese from user-authored "
+        "text; do not collapse them into generic Chinese."
     )
 
 
@@ -526,6 +599,7 @@ def dag_step_language_rules(*, subject: str = "output language policy") -> str:
 
 OutputLanguageSection = Literal[
     "root_system_context",
+    "root_existing_request",
     "dag_step_scope",
     "dag_step_rules",
     "dag_step_request_anchor",
@@ -551,7 +625,11 @@ def output_language_directives(
         # beside it would hand the model a second, competing rule.
         if language:
             return f"Output language policy:\n{output_language_policy(language)}"
-        return response_language_rules()
+        return request_only_language_harness(request)
+    if section == "root_existing_request":
+        if language:
+            return f"Output language policy:\n{output_language_policy(language)}"
+        return _root_request_language_policy()
     if section == "dag_step_scope":
         return output_language_policy(language).strip()
     if section == "dag_step_rules":
@@ -563,11 +641,13 @@ def output_language_directives(
             return ""
         # Quoted whole: any truncation can drop an explicit target-language
         # instruction sitting in the middle of a long request.
-        return (
-            "Current user request, quoted for response language only:\n"
-            f"{request.strip()}\n\n"
-            "This request is not the executable goal for this step; use it "
-            "only to decide the natural language of user-facing prose.\n\n"
-            f"{response_language_rules()}"
-        )
-    return output_language_policy(language)
+        return request_only_language_harness(request)
+    if language:
+        return output_language_policy(language)
+    if section == "dag_step_instruction":
+        return _dag_step_instruction_language_policy()
+    if section == "plan_payload":
+        return _structured_request_language_policy("latest_user_request")
+    if section == "completion_assessment":
+        return _structured_request_language_policy("user_authored_language_request")
+    return request_only_language_harness(request)

src/xagent/core/agent/pattern/dag/dag.py

@@ -1503,6 +1503,8 @@ async def _assess_completion(
         return assessment
 
     def _completion_assessment_messages(self, context: Any) -> list[dict[str, Any]]:
+        language_request = latest_user_text(context, prefer_display=True) or ""
+        output_language = effective_output_language(context)
         latest_messages = [
             {"role": message.role, "content": message.content}
             for message in getattr(context, "messages", [])
@@ -1515,8 +1517,14 @@ def _completion_assessment_messages(self, context: Any) -> list[dict[str, Any]]:
         ]
         payload = {
             "output_language_policy": output_language_directives(
-                effective_output_language(context),
+                output_language,
                 section="completion_assessment",
+                request=language_request,
+            ),
+            **(
+                {}
+                if output_language
+                else {"user_authored_language_request": language_request}
             ),
             "authoritative_user_requests": authoritative_user_requests,
             "messages": latest_messages,

src/xagent/core/agent/pattern/dag/plan_generator.py

@@ -579,6 +579,7 @@ def _build_prompt(self, request: PlanGenerationRequest) -> str:
                 if language_source == OUTPUT_LANGUAGE_SOURCE_PLAN
                 else expected_language,
                 section="plan_payload",
+                request=latest_request,
             ),
             "messages": latest_messages,
             "retrieved_memory_context": request.context.metadata.get(

src/xagent/core/agent/pattern/react/react.py

@@ -132,10 +132,9 @@ class ReActReasoningMode(str, Enum):
 REACT_RESPONSE_LANGUAGE_DESCRIPTION = (
     "Target natural language for user-facing prose in this ReAct response, "
     "for example English, Simplified Chinese, Traditional Chinese, or Spanish. "
+    "Follow the authoritative output language guidance in the system context. "
     "For Chinese requests, choose Simplified Chinese or Traditional Chinese to "
-    "match the request script; do not use generic Chinese. If the current user "
-    "request explicitly asks to answer in another language, use that requested "
-    "target language."
+    "match the request script; do not use generic Chinese."
 )
 
 

tests/core/agent/test_auto.py

@@ -25,7 +25,7 @@
     OUTPUT_LANGUAGE_METADATA_KEY,
     OUTPUT_LANGUAGE_SOURCE_METADATA_KEY,
     OUTPUT_LANGUAGE_SOURCE_PLAN,
-    response_language_rules,
+    output_language_directives,
 )
 from xagent.core.agent.pattern.auto.auto import DECISION_TOOL_NAME, _AutoChildRuntime
 from xagent.core.model.chat.basic.router import RouterLLM
@@ -857,10 +857,14 @@ async def test_auto_pattern_final_answer_completes_without_child_pattern() -> No
     assert runtime.last_checkpoint is not None
     assert runtime.last_checkpoint["pattern"] == "AutoPattern"
     assert (
-        "same natural language as the current user request"
+        "authoritative output language guidance in the system context"
         in tool_schema["description"]
     )
-    assert "tool results, source documents" in answer_schema["description"]
+    assert "connector metadata" in answer_schema["description"]
+    assert (
+        output_language_directives("", section="root_existing_request")
+        in llm.calls[0]["messages"][0]["content"]
+    )
 
 
 @pytest.mark.asyncio
@@ -2293,7 +2297,9 @@ async def test_stale_memory_language_does_not_reach_child_as_hard_policy() -> No
     assert "Output language:" not in child_system
     assert "Output language policy:" not in child_system
     assert "Summarize the quarterly revenue trend in one paragraph." in child_system
-    assert response_language_rules() in child_system
+    assert (
+        output_language_directives("", section="root_existing_request") in child_system
+    )
 
 
 @pytest.mark.asyncio
@@ -2319,18 +2325,18 @@ async def test_direct_final_answer_allows_an_explicit_target_language() -> None:
     assert result["success"] is True
     assert result["output"] == "La capitale de l'Italie est Rome."
     assert OUTPUT_LANGUAGE_METADATA_KEY not in context.metadata
-    target_rule = (
-        "If the current user request explicitly asks to translate, rewrite, or "
-        "answer in another language, use that requested target language."
-    )
+    target_rule = "Honor any explicit or implicit request to translate"
     tool_schema = llm.calls[0]["tools"][0]["function"]
     assert target_rule in tool_schema["description"]
     assert (
         target_rule in tool_schema["parameters"]["properties"]["answer"]["description"]
     )
     system_content = context.get_messages_for_llm()[0]["content"]
-    assert request in system_content
-    assert target_rule in system_content
+    assert system_content.count(request) == 1
+    assert (
+        output_language_directives("", section="root_existing_request")
+        in system_content
+    )
 
 
 class RoutedDecisionLLM: