ref(agent): canonicalize request language and pending answers (toby)

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

@@ -35,6 +35,42 @@ class TopLevelUserRequest:
     has_pending_response: bool = False
 
 
+@dataclass(frozen=True)
+class PendingUserResponse:
+    """Allowlisted context for one answer to a pending agent message."""
+
+    answer: str
+    question: str
+    message_type: str
+
+
+def pending_user_response(message: Any) -> PendingUserResponse | None:
+    """Extract only language-relevant fields from a marked user message."""
+    if getattr(message, "role", None) != "user":
+        return None
+    metadata = getattr(message, "metadata", None)
+    marker = (
+        metadata.get("response_to_waiting_for_user")
+        if isinstance(metadata, dict)
+        else None
+    )
+    if not isinstance(marker, dict):
+        return None
+    question = marker.get("question")
+    if not isinstance(question, str) or not question.strip():
+        return None
+    raw_message_type = marker.get("message_type", "question")
+    message_type = (
+        raw_message_type.strip()
+        if isinstance(raw_message_type, str) and raw_message_type.strip()
+        else "question"
+    )
+    answer = getattr(message, "content", "")
+    if not isinstance(answer, str):
+        return None
+    return PendingUserResponse(answer, question, message_type)
+
+
 def _stored_top_level_user_request(context: Any) -> TopLevelUserRequest | None:
     metadata = getattr(context, "metadata", None)
     if not isinstance(metadata, dict):
@@ -193,7 +229,7 @@ def top_level_user_request(context: Any) -> TopLevelUserRequest:
             continue
         metadata = getattr(message, "metadata", None)
         metadata = metadata if isinstance(metadata, dict) else {}
-        if metadata.get("response_to_waiting_for_user"):
+        if pending_user_response(message) is not None:
             has_pending_response = True
             continue
         if metadata.get("dag_step_id"):

src/xagent/core/agent/language.py

@@ -1,10 +1,13 @@
 """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
 
+from .context.enrichment import PendingUserResponse, TopLevelUserRequest
+
 OUTPUT_LANGUAGE_METADATA_KEY = "output_language"
 OUTPUT_LANGUAGE_SOURCE_METADATA_KEY = "output_language_source"
 OUTPUT_LANGUAGE_SOURCE_PLAN = "dag_plan"
@@ -190,6 +193,54 @@ class ResponseLanguageScriptMismatch:
     latin_count: int
 
 
+def serialize_pending_user_response(response: PendingUserResponse) -> dict[str, str]:
+    """Serialize the allowlisted pending-response fields without truncation."""
+    return {
+        "answer": response.answer,
+        "question": response.question,
+        "message_type": response.message_type,
+    }
+
+
+def canonical_unpinned_request_language_policy() -> str:
+    """Return the canonical soft-authority policy for future consumers."""
+    return (
+        "A caller-provided request_context.output_language is the sole hard "
+        "language authority. When it is absent, use independent_user_request as "
+        "the baseline for the language and script of user-facing prose. Honor its "
+        "explicit or implicit target-language intent, including requests to "
+        "translate or rewrite content for another-language audience. A "
+        "pending_response may override that baseline only when its answer "
+        "explicitly asks to translate, rewrite, or continue in another language, "
+        "or when its question explicitly asks for the output language or script "
+        "and its answer is an unambiguous selection. A language name is not an "
+        "override when the pending question asks for another kind of value; for "
+        'example, "Which city should the email mention?" followed by "Spanish" '
+        "remains ordinary conversation context. Names, addresses, connector "
+        "metadata, tool results, sources, memory, examples, DAG text, and "
+        "dependency results are not language evidence. Preserve Simplified "
+        "Chinese versus Traditional Chinese. This policy controls language only "
+        "and never replaces or narrows the executable request."
+    )
+
+
+def render_request_language_harness(
+    request: TopLevelUserRequest,
+    pending_response: PendingUserResponse | None = None,
+) -> str:
+    """Render exact request-only evidence for future language consumers."""
+    evidence: dict[str, Any] = {
+        "independent_user_request": request.language_text,
+    }
+    if pending_response is not None:
+        evidence["pending_response"] = serialize_pending_user_response(pending_response)
+    return (
+        "Canonical request-language evidence (JSON):\n"
+        f"{json.dumps(evidence, ensure_ascii=False)}\n\n"
+        f"{canonical_unpinned_request_language_policy()}"
+    )
+
+
 def _script_counts(prose: str) -> tuple[int, int]:
     prose_without_technical_spans = _TECHNICAL_SPAN_PATTERN.sub(" ", prose)
     han_count = sum(

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

@@ -1886,11 +1886,30 @@ def _forward_user_response_to_waiting_step(self, root_context: Any) -> bool:
 
         child_context = type(root_context).from_dict(active_context)
         self._refresh_restored_step_runtime_metadata(child_context, root_context)
+        state = self.active_step_pattern_states.get(step_id)
+        waiting_request = (
+            state.get("waiting_for_user_request") if isinstance(state, dict) else {}
+        )
+        waiting_request = waiting_request if isinstance(waiting_request, dict) else {}
         for message in root_user_messages[self.planned_user_message_count :]:
+            marker = {
+                "question": waiting_request.get("message", ""),
+                "message_type": waiting_request.get("message_type", "question"),
+            }
+            metadata = {
+                **getattr(message, "metadata", {}),
+                "response_to_waiting_for_user": marker,
+            }
+            root_index = next(
+                index
+                for index, root_message in enumerate(root_context.messages)
+                if root_message is message
+            )
+            root_context.messages[root_index] = replace(message, metadata=metadata)
             child_context.add_user_message(
                 message.content,
                 metadata={
-                    **getattr(message, "metadata", {}),
+                    **metadata,
                     "kind": "dag_waiting_user_response",
                     "forwarded_from_root": True,
                     "dag_step_id": step_id,

tests/core/agent/test_request_language_policy.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from xagent.core.agent.context import ExecutionContext
+from xagent.core.agent.context.enrichment import (
+    PendingUserResponse,
+    TopLevelUserRequest,
+    pending_user_response,
+    top_level_user_request,
+)
+from xagent.core.agent.language import (
+    canonical_unpinned_request_language_policy,
+    render_request_language_harness,
+    serialize_pending_user_response,
+)
+from xagent.core.agent.pattern.dag.dag import DAGPattern
+
+
+def _request(text: str) -> TopLevelUserRequest:
+    return TopLevelUserRequest(text, text, "text")
+
+
+def _marked_message(answer: str, marker: Any) -> Any:
+    context = ExecutionContext()
+    return context.add_user_message(
+        answer,
+        metadata={"response_to_waiting_for_user": marker},
+    )
+
+
+def test_pending_response_serializer_exposes_only_allowlisted_exact_fields() -> None:
+    answer = "ANSWER_BEGIN_" + "答" * 8_000 + "_ANSWER_END"
+    question = "Which output language? " + "Q" * 8_000
+    message = _marked_message(
+        answer,
+        {
+            "question": question,
+            "message_type": "question",
+            "tool_name": "private_connector",
+            "tool_call_id": "secret-id",
+            "interactions": [{"options": ["Spanish"]}],
+            "requests": [{"internal": True}],
+        },
+    )
+
+    response = pending_user_response(message)
+    assert response is not None
+    serialized = serialize_pending_user_response(response)
+
+    assert serialized == {
+        "answer": answer,
+        "question": question,
+        "message_type": "question",
+    }
+    serialized_text = json.dumps(serialized, ensure_ascii=False)
+    assert serialized_text.count(answer) == 1
+    assert serialized_text.count(question) == 1
+    assert "private_connector" not in serialized_text
+    assert "secret-id" not in serialized_text
+    assert "options" not in serialized_text
+
+
+@pytest.mark.parametrize(
+    "marker",
+    [True, False, "legacy", 1, None, {}, {"question": " \n"}],
+)
+def test_pending_response_malformed_or_blank_marker_degrades_safely(
+    marker: Any,
+) -> None:
+    message = _marked_message("Spanish", marker)
+    assert pending_user_response(message) is None
+    context = SimpleNamespace(messages=[message], metadata={})
+    assert top_level_user_request(context).language_text == "Spanish"
+
+
+def test_language_question_and_terse_selection_are_preserved_for_policy() -> None:
+    response = pending_user_response(
+        _marked_message(
+            "Spanish",
+            {"question": "Which output language?", "message_type": "question"},
+        )
+    )
+    assert response is not None
+    harness = render_request_language_harness(_request("Draft the email."), response)
+    evidence = json.loads(harness.split("\n", 2)[1])
+
+    assert evidence["pending_response"] == {
+        "answer": "Spanish",
+        "question": "Which output language?",
+        "message_type": "question",
+    }
+    assert "question explicitly asks for the output language or script" in harness
+    assert "answer is an unambiguous selection" in harness
+
+
+def test_explicit_answer_override_and_city_negative_control_share_one_policy() -> None:
+    policy = canonical_unpinned_request_language_policy()
+
+    assert "request_context.output_language is the sole hard language authority" in (
+        policy
+    )
+    assert "answer explicitly asks to translate, rewrite, or continue" in policy
+    assert '"Which city should the email mention?" followed by "Spanish"' in policy
+    assert "remains ordinary conversation context" in policy
+
+
+def test_harness_preserves_large_request_and_answer_exactly_once() -> None:
+    request = "REQUEST_BEGIN_" + "請" * 8_000 + "_REQUEST_END"
+    answer = "Continue in Spanish. " + "A" * 8_000
+    response = PendingUserResponse(answer, "Which language?", "question")
+
+    harness = render_request_language_harness(_request(request), response)
+
+    assert harness.count(request) == 1
+    assert harness.count(answer) == 1
+    assert harness.count("Which language?") == 1
+
+
+def test_new_policy_is_not_active_in_existing_consumers() -> None:
+    context = ExecutionContext()
+    context.add_user_message("Draft the email.")
+
+    assert "Canonical request-language evidence" not in context._system_context()
+    assert "Canonical request-language evidence" not in json.dumps(
+        context.get_messages_for_llm()
+    )
+
+
+def test_dag_marker_is_propagated_symmetrically_without_internal_fields() -> None:
+    root = ExecutionContext()
+    root.add_user_message("Draft the email.")
+    child = root.create_child_context(execution_id="step")
+    pattern = DAGPattern(lambda **_: None)
+    pattern.status = "waiting_for_user"
+    pattern.active_step_id = "draft"
+    pattern.active_step_ids = ["draft"]
+    pattern.active_step_contexts = {"draft": child.to_dict()}
+    pattern.active_step_pattern_states = {
+        "draft": {
+            "status": "waiting_for_user",
+            "waiting_for_user_request": {
+                "message": "Which output language?",
+                "message_type": "question",
+                "tool_call_id": "secret-id",
+            },
+        }
+    }
+    pattern.planned_user_message_count = 1
+    root.add_user_message("Spanish")
+
+    assert pattern._forward_user_response_to_waiting_step(root)
+    expected = {
+        "question": "Which output language?",
+        "message_type": "question",
+    }
+    assert root.messages[-1].metadata["response_to_waiting_for_user"] == expected
+    restored_child = ExecutionContext.from_dict(pattern.active_step_contexts["draft"])
+    assert (
+        restored_child.messages[-1].metadata["response_to_waiting_for_user"] == expected
+    )