From 04a9833763f7daae9b3ff19d9c1d45fb6ab21c61 Mon Sep 17 00:00:00 2001 From: 2SinceJuly <2929362966@qq.com> Date: Thu, 12 Nov 2026 22:30:48 +0800 Subject: [PATCH] fix(subagents): keep a steered subagent report in the parent snapshot (#8167) * fix(subagents): keep a steered subagent report in the parent snapshot A Steer frees the parent's turn before its subagent finishes: wait_tool returns a 'result is pending' task result and the real report arrives later as a hidden subagent-result HumanMessage. ParentContextSnapshot only kept genuine user messages, so it dropped that report and kept the pending placeholder. A later context_mode=snapshot delegation therefore inherited a pending result and could not use quotes the parent had already received. Keep the hidden subagent-result message as historical conversation data. * fix(subagents): share the delayed result message name --------- Co-authored-by: 2SinceJuly <202121119+2SinceJuly@users.noreply.github.com> Co-authored-by: AochenShen99 Co-authored-by: Willem Jiang --- .../deerflow/runtime/runs/input_control.py | 4 ++- .../harness/deerflow/subagents/AGENTS.md | 3 +++ .../deerflow/subagents/context_snapshot.py | 23 ++++++++++++---- .../harness/deerflow/utils/messages.py | 1 + backend/tests/test_run_input_graph.py | 6 +++++ .../tests/test_subagent_context_snapshot.py | 27 ++++++++++++++++++- 6 files changed, 57 insertions(+), 7 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/runs/input_control.py b/backend/packages/harness/deerflow/runtime/runs/input_control.py index fe6e836663..90e9af436c 100644 --- a/backend/packages/harness/deerflow/runtime/runs/input_control.py +++ b/backend/packages/harness/deerflow/runtime/runs/input_control.py @@ -9,6 +9,8 @@ from uuid import uuid4 from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from deerflow.utils.messages import SUBAGENT_RESULT_MESSAGE_NAME + from .inputs import RunInputs from .schemas import GOAL_RUN_HARD_STOP_REASONS @@ -223,7 +225,7 @@ class RunInputControl: self.receipts.append( HumanMessage( content=json.dumps({"source": "subagent", "task_call_id": call_id, "result": message.content}, ensure_ascii=False), - name="subagent-result", + name=SUBAGENT_RESULT_MESSAGE_NAME, id="run-input-result-" + self.inputs.run_id + "-" + call_id, additional_kwargs={"hide_from_ui": True, "untrusted_input": True, "origin_task_id": call_id, "run_id": self.inputs.run_id}, ) diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 293d0d84cb..9f996dd45f 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -18,6 +18,9 @@ make test-blocking-io Batch and result-reader tests run against SQLite and, when `TEST_POSTGRES_URI` is set, against an isolated PostgreSQL schema. They need no paid model. +Run input delivery and parent context snapshots share `SUBAGENT_RESULT_MESSAGE_NAME` +from `deerflow.utils.messages`. Keep delayed Steer reports as sanitized history. + ## Terminal state - **The first terminal write wins.** Cancellation, timeout and completion race. diff --git a/backend/packages/harness/deerflow/subagents/context_snapshot.py b/backend/packages/harness/deerflow/subagents/context_snapshot.py index 4456529aef..39904eea57 100644 --- a/backend/packages/harness/deerflow/subagents/context_snapshot.py +++ b/backend/packages/harness/deerflow/subagents/context_snapshot.py @@ -13,6 +13,7 @@ from langchain_core.messages import AIMessage, HumanMessage, ToolMessage, conver from deerflow.agents.middlewares.input_sanitization_middleware import neutralize_untrusted_tags from deerflow.agents.middlewares.message_utils import is_genuine_user_message +from deerflow.utils.messages import SUBAGENT_RESULT_MESSAGE_NAME logger = logging.getLogger(__name__) @@ -84,10 +85,22 @@ def _is_conversation_message(message: Any) -> bool: if isinstance(message, HumanMessage): # Hidden clarification responses are user input; memory/todo reminders # and other framework-injected HumanMessages are not. - return is_genuine_user_message(message) + return is_genuine_user_message(message) or _is_steered_subagent_report(message) return isinstance(message, (AIMessage, ToolMessage)) and not message.additional_kwargs.get("hide_from_ui") +def _is_steered_subagent_report(message: HumanMessage) -> bool: + """Return True for the hidden report a Steer delivers after freeing the turn. + + ``RunInputControl.wait_tool`` ends the parent turn with a "result is + pending" ``task`` result and appends the real report later as this hidden + message. It is framework-delivered conversation data, not parent authority: + the next delegation must inherit it, because the visible ``task`` result + stays pending in the retained history. + """ + return message.name == SUBAGENT_RESULT_MESSAGE_NAME and bool(message.content) + + @dataclass(frozen=True) class ParentContextSnapshot: """Serialized content has no aliases to parent state or sibling executions.""" @@ -100,10 +113,10 @@ class ParentContextSnapshot: A single background HumanMessage avoids replaying parent tool protocol frames into child receipts, step events, skill policy, or turn budgets. - Runtime state, parent system instructions, hidden framework messages, - artifacts and message metadata never cross this boundary. No extra - truncation hides retained history; the caller opts into its input-token - cost and normal child compaction. + Runtime state, parent system instructions, unrelated hidden framework + messages, artifacts and message metadata never cross this boundary. + No extra truncation hides retained history; the caller opts into its + input-token cost and normal child compaction. """ blocks: list[dict[str, Any]] = [] diff --git a/backend/packages/harness/deerflow/utils/messages.py b/backend/packages/harness/deerflow/utils/messages.py index 10befa1331..deb7c86ac2 100644 --- a/backend/packages/harness/deerflow/utils/messages.py +++ b/backend/packages/harness/deerflow/utils/messages.py @@ -8,6 +8,7 @@ from langchain_core.messages import HumanMessage ORIGINAL_USER_CONTENT_KEY = "original_user_content" SUMMARY_MESSAGE_NAME = "summary" +SUBAGENT_RESULT_MESSAGE_NAME = "subagent-result" #: Server-owned mark the Gateway stamps on an untrusted caller's message when it #: carries a framework marker (``hide_from_ui``, a ``summary`` name) that would diff --git a/backend/tests/test_run_input_graph.py b/backend/tests/test_run_input_graph.py index ce72943202..a0e9bae7dc 100644 --- a/backend/tests/test_run_input_graph.py +++ b/backend/tests/test_run_input_graph.py @@ -22,7 +22,9 @@ from deerflow.config.token_budget_config import TokenBudgetConfig from deerflow.runtime.runs.input_control import ChildInputGate, RunInputControl from deerflow.runtime.runs.inputs import INPUT_CONTEXT_KEY, RunInputs from deerflow.runtime.runs.store.memory import MemoryRunStore +from deerflow.subagents.context_snapshot import ParentContextSnapshot from deerflow.subagents.status_contract import SUBAGENT_TOKEN_USAGE_KEY +from deerflow.utils.messages import message_content_to_text def test_input_control_reaches_the_factory_only_from_the_worker(): @@ -497,6 +499,10 @@ async def test_graph_applies_input_before_long_tool_finishes_and_confirms_durabl release.set() result = await asyncio.wait_for(run, 5) assert result["applied_run_inputs"] == {"run_id": "run", "ids": ["extra"]} + assert any(isinstance(message, ToolMessage) and message.additional_kwargs.get("subagent_pending") for message in result["messages"]) + snapshot = ParentContextSnapshot.from_state(result) + assert snapshot is not None + assert "Long operation completed" in message_content_to_text(snapshot.to_message().content) # Summaries can remove message text; the independent graph receipt # still proves it was applied before the store acknowledgement. ids = [m.id for m in result["messages"] if m.additional_kwargs.get("run_input_id")] diff --git a/backend/tests/test_subagent_context_snapshot.py b/backend/tests/test_subagent_context_snapshot.py index db3cfbddf4..0ec2dec3d3 100644 --- a/backend/tests/test_subagent_context_snapshot.py +++ b/backend/tests/test_subagent_context_snapshot.py @@ -8,7 +8,7 @@ import pytest from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage from deerflow.subagents.context_snapshot import ParentContextSnapshot -from deerflow.utils.messages import message_content_to_text +from deerflow.utils.messages import SUBAGENT_RESULT_MESSAGE_NAME, message_content_to_text @pytest.mark.parametrize("dispatch_id", ["dispatch", "earlier-delegation"]) @@ -346,6 +346,31 @@ def test_snapshot_keeps_only_valid_hidden_user_responses(response_kind): assert "" not in snapshot.content_json +def test_snapshot_keeps_a_steered_subagent_report_the_parent_already_has(): + """A Steer frees the parent's turn before its subagent finishes. + + ``RunInputControl.wait_tool`` returns a "result is pending" ``task`` + placeholder and delivers the real report later as a hidden + ``subagent-result`` HumanMessage. The snapshot dropped that hidden message + and kept the placeholder, so the next delegation inherited a pending call + instead of the report the parent had already received. + """ + call = {"name": "task", "args": {"prompt": "read the quotes"}, "id": "call-1"} + report = json.dumps({"source": "subagent", "task_call_id": "call-1", "result": "North Paper 38.75, West Paper 42.50"}, ensure_ascii=False) + messages = [ + HumanMessage(content="Delegate to general-purpose", id="user-1"), + AIMessage(content="", tool_calls=[call], id="ai-1"), + ToolMessage(content="The subagent result is pending and will arrive in this Run. The user has sent an updated requirement.", tool_call_id="call-1", name="task", id="pending-1", additional_kwargs={"subagent_pending": True}), + HumanMessage(content=report, name=SUBAGENT_RESULT_MESSAGE_NAME, id="run-input-result-run-1-call-1", additional_kwargs={"hide_from_ui": True, "untrusted_input": True}), + HumanMessage(content="Ask the reviewer to compare the quotes", id="user-2"), + ] + + snapshot = ParentContextSnapshot.from_state({"messages": messages}) + text = message_content_to_text(snapshot.to_message().content) + + assert "North Paper 38.75, West Paper 42.50" in text + + @pytest.mark.parametrize("injection", ["legacy-summary", "previous-snapshot"]) def test_framework_only_history_has_no_snapshot(injection): message = HumanMessage(content="PRIVATE_SUMMARY", name="summary") if injection == "legacy-summary" else ParentContextSnapshot.from_state({"messages": [HumanMessage(content="PRIVATE_ANCESTOR_CONTEXT")]}).to_message()