mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-12 22:20:33 +00:00
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 <sac1063467113@hotmail.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
bea75f3b23
commit
04a9833763
@ -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},
|
||||
)
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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]] = []
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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")]
|
||||
|
||||
@ -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 "<system>" 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()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user