From 63ddcd2254070ec0746b21701d7dd511598ecbcf Mon Sep 17 00:00:00 2001 From: OliverBryant <2713999266@qq.com> Date: Thu, 3 Sep 2026 14:58:34 +0800 Subject: [PATCH 1/3] ref(agent): canonicalize request language and pending answers --- src/xagent/core/agent/context/enrichment.py | 38 +++- src/xagent/core/agent/language.py | 51 ++++++ src/xagent/core/agent/pattern/dag/dag.py | 21 ++- .../agent/test_request_language_policy.py | 165 ++++++++++++++++++ 4 files changed, 273 insertions(+), 2 deletions(-) create mode 100644 tests/core/agent/test_request_language_policy.py diff --git a/src/xagent/core/agent/context/enrichment.py b/src/xagent/core/agent/context/enrichment.py index 4edc3dee1e..17cb4886d5 100644 --- a/src/xagent/core/agent/context/enrichment.py +++ b/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"): diff --git a/src/xagent/core/agent/language.py b/src/xagent/core/agent/language.py index a3ae5fd346..5108d61b1d 100644 --- a/src/xagent/core/agent/language.py +++ b/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( diff --git a/src/xagent/core/agent/pattern/dag/dag.py b/src/xagent/core/agent/pattern/dag/dag.py index 8adf709d6a..7526ad4d68 100644 --- a/src/xagent/core/agent/pattern/dag/dag.py +++ b/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, diff --git a/tests/core/agent/test_request_language_policy.py b/tests/core/agent/test_request_language_policy.py new file mode 100644 index 0000000000..52c62cbe96 --- /dev/null +++ b/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 + ) From d7096e577427e08b80699c5bd77fbb9ac02158a9 Mon Sep 17 00:00:00 2001 From: OliverBryant <2713999266@qq.com> Date: Thu, 3 Sep 2026 15:14:17 +0800 Subject: [PATCH 2/3] test(agent): align DAG pending response metadata --- tests/core/agent/test_dag.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/core/agent/test_dag.py b/tests/core/agent/test_dag.py index a7482e08b3..d916432192 100644 --- a/tests/core/agent/test_dag.py +++ b/tests/core/agent/test_dag.py @@ -454,6 +454,10 @@ def test_dag_waiting_response_preserves_active_step_state() -> None: "kind": "dag_waiting_user_response", "forwarded_from_root": True, "dag_step_id": "confirm", + "response_to_waiting_for_user": { + "question": "Choose A or B", + "message_type": "question", + }, } From c4ef6ddd187152b830efa9dc12710f88dafc76e9 Mon Sep 17 00:00:00 2001 From: OliverBryant <2713999266@qq.com> Date: Thu, 3 Sep 2026 15:58:45 +0800 Subject: [PATCH 3/3] fix(agent): keep request language layer behavior-neutral --- src/xagent/core/agent/context/enrichment.py | 2 +- src/xagent/core/agent/pattern/dag/dag.py | 21 +----- tests/core/agent/test_dag.py | 4 - .../agent/test_request_language_policy.py | 74 ++++++++++++++----- 4 files changed, 59 insertions(+), 42 deletions(-) diff --git a/src/xagent/core/agent/context/enrichment.py b/src/xagent/core/agent/context/enrichment.py index 17cb4886d5..48fd277ee9 100644 --- a/src/xagent/core/agent/context/enrichment.py +++ b/src/xagent/core/agent/context/enrichment.py @@ -229,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 pending_user_response(message) is not None: + if metadata.get("response_to_waiting_for_user"): has_pending_response = True continue if metadata.get("dag_step_id"): diff --git a/src/xagent/core/agent/pattern/dag/dag.py b/src/xagent/core/agent/pattern/dag/dag.py index 7526ad4d68..8adf709d6a 100644 --- a/src/xagent/core/agent/pattern/dag/dag.py +++ b/src/xagent/core/agent/pattern/dag/dag.py @@ -1886,30 +1886,11 @@ 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={ - **metadata, + **getattr(message, "metadata", {}), "kind": "dag_waiting_user_response", "forwarded_from_root": True, "dag_step_id": step_id, diff --git a/tests/core/agent/test_dag.py b/tests/core/agent/test_dag.py index d916432192..a7482e08b3 100644 --- a/tests/core/agent/test_dag.py +++ b/tests/core/agent/test_dag.py @@ -454,10 +454,6 @@ def test_dag_waiting_response_preserves_active_step_state() -> None: "kind": "dag_waiting_user_response", "forwarded_from_root": True, "dag_step_id": "confirm", - "response_to_waiting_for_user": { - "question": "Choose A or B", - "message_type": "question", - }, } diff --git a/tests/core/agent/test_request_language_policy.py b/tests/core/agent/test_request_language_policy.py index 52c62cbe96..6d106f0e56 100644 --- a/tests/core/agent/test_request_language_policy.py +++ b/tests/core/agent/test_request_language_policy.py @@ -1,7 +1,6 @@ from __future__ import annotations import json -from types import SimpleNamespace from typing import Any import pytest @@ -65,17 +64,59 @@ def test_pending_response_serializer_exposes_only_allowlisted_exact_fields() -> assert "options" not in serialized_text +@pytest.mark.parametrize( + "metadata", + [ + None, + {"response_to_waiting_for_user": True}, + {"response_to_waiting_for_user": False}, + {"response_to_waiting_for_user": "legacy"}, + {"response_to_waiting_for_user": 1}, + {"response_to_waiting_for_user": None}, + {"response_to_waiting_for_user": {}}, + {"response_to_waiting_for_user": {"question": " \n"}}, + {"response_to_waiting_for_user": {"question": 7}}, + ], +) +def test_pending_response_rejects_malformed_or_blank_marker( + metadata: dict[str, Any] | None, +) -> None: + context = ExecutionContext() + message = context.add_user_message("Spanish", metadata=metadata) + assert pending_user_response(message) is None + + +def test_pending_response_defaults_invalid_message_type_without_leaking_it() -> None: + response = pending_user_response( + _marked_message( + "Spanish", + {"question": "Which output language?", "message_type": ["internal"]}, + ) + ) + + assert response == PendingUserResponse( + answer="Spanish", + question="Which output language?", + message_type="question", + ) + + @pytest.mark.parametrize( "marker", - [True, False, "legacy", 1, None, {}, {"question": " \n"}], + [True, "legacy", {"question": " \n"}, {"question": 7}], ) -def test_pending_response_malformed_or_blank_marker_degrades_safely( +def test_strict_parser_does_not_change_layer_a_marker_compatibility( marker: Any, ) -> None: - message = _marked_message("Spanish", marker) + context = ExecutionContext() + context.add_user_message("Draft the email.") + message = context.add_user_message( + "Spanish", + metadata={"response_to_waiting_for_user": marker}, + ) + assert pending_user_response(message) is None - context = SimpleNamespace(messages=[message], metadata={}) - assert top_level_user_request(context).language_text == "Spanish" + assert top_level_user_request(context).language_text == "Draft the email." def test_language_question_and_terse_selection_are_preserved_for_policy() -> None: @@ -98,13 +139,18 @@ def test_language_question_and_terse_selection_are_preserved_for_policy() -> Non assert "answer is an unambiguous selection" in harness -def test_explicit_answer_override_and_city_negative_control_share_one_policy() -> None: +def test_caller_pin_and_explicit_answer_override_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 + + +def test_city_question_and_language_name_are_not_a_language_override() -> None: + policy = canonical_unpinned_request_language_policy() + assert '"Which city should the email mention?" followed by "Spanish"' in policy assert "remains ordinary conversation context" in policy @@ -121,7 +167,7 @@ def test_harness_preserves_large_request_and_answer_exactly_once() -> None: assert harness.count("Which language?") == 1 -def test_new_policy_is_not_active_in_existing_consumers() -> None: +def test_request_language_harness_is_not_active_in_root_consumers() -> None: context = ExecutionContext() context.add_user_message("Draft the email.") @@ -131,7 +177,7 @@ def test_new_policy_is_not_active_in_existing_consumers() -> None: ) -def test_dag_marker_is_propagated_symmetrically_without_internal_fields() -> None: +def test_request_language_representation_is_not_active_in_dag_forwarding() -> None: root = ExecutionContext() root.add_user_message("Draft the email.") child = root.create_child_context(execution_id="step") @@ -154,12 +200,6 @@ def test_dag_marker_is_propagated_symmetrically_without_internal_fields() -> Non 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 + assert "response_to_waiting_for_user" not in root.messages[-1].metadata restored_child = ExecutionContext.from_dict(pattern.active_step_contexts["draft"]) - assert ( - restored_child.messages[-1].metadata["response_to_waiting_for_user"] == expected - ) + assert "response_to_waiting_for_user" not in restored_child.messages[-1].metadata