github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-11-12 14:00:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(subagents): keep tests_passed evidence per call when a provider reuses its id (#8012)

Browse Source 
* fix(subagents): keep tests_passed evidence per call when a provider reuses its id

Tool-call ids are unique per response, not per conversation. Some providers
number calls per response, so two turns can both issue call_0.

The bash evidence for tests_passed acceptance assumed one call per id in two
places:

- _harvest_bash_executions built one id -> command map over the whole
  conversation, so every result under a reused id was labelled with the
  last call's command. A passing subset run (pytest tests/unit) became
  evidence for the full suite when a later call_0 ran pytest tests and had
  no result yet.
- update_bash_executions merged values chunks by tool_call_id. A dict keeps
  a key's first position, so a later run under a reused id overwrote the
  earlier entry in place. With call_0 failing, call_1 passing and call_0
  failing again, the merged list ended on the stale pass and the leaf held.

Pair results through pair_tool_call_results, which answers a call only from
its own response, and merge chunks by result message id, keeping the call id
for entries without one. Each run stays a separate entry in execution order,
so the latest run decides the leaf again.

* fix(subagents): key the summary cache per run and document id-less calls

- Cache full-output test summaries by result message id (call id only
  without one), so reused call ids keep one cache slot per run.
- Update the bash_executions field doc and the subagents guide to the
  result-id merge, and state that calls without an id record no
  evidence (the leaf stays UNVERIFIED).
- Tests pin both: one classification per run under a reused id, and no
  evidence for a call with a missing or empty id.

* chore(subagents): drop the stray empty deerflow/executor.py

The previous commit added a 0-byte module at the package root by mistake.
Nothing imports deerflow.executor; the real module is deerflow/subagents/executor.py.
This commit is contained in:
Evan 2026-11-09 06:59:47 -05:00 committed by GitHub
parent fb094984e3
commit 3e59e22b4c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 232 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@ -2926,6 +2926,11 @@ independently shaped summaries from other runners still contribute evidence.
Pytest sibling summaries retain warning counts and the optional human-readable
duration suffix printed for runs lasting at least a minute, such as `(0:01:05)`.
`tests_passed` evidence pairs each command with its own result, even when a
provider reuses a tool-call id in a later response. Each run stays a separate
entry in execution order, so a later failing run is decisive and an earlier
run's output is never credited to another command.
For Maven and Gradle test acceptance, a `BUILD SUCCESS` / `BUILD SUCCESSFUL`
banner alone remains `UNVERIFIED`: builds can succeed without running tests.
Complete `Tests run: N, Failures: F, Errors: E, Skipped: S` or `N tests completed`

7
backend/packages/harness/deerflow/subagents/AGENTS.md
View File

File diff suppressed because one or more lines are too long

78
backend/packages/harness/deerflow/subagents/executor.py
View File

@ -155,7 +155,8 @@ class SubagentResult:
``persistent_shell_sessions`` flag resolved from the state that
carried the evidence (``None`` when unidentifiable — the matcher
fails closed on it).
Accumulated per chunk (merged by ``tool_call_id``) so
Accumulated per chunk (merged by result message id, ``tool_call_id``
only for entries without one) so
summarization compacting earlier messages cannot erase a recorded
execution. ``None`` when the delegation carried no acceptance
criteria, the run ended before streaming, or harvesting failed;
@ -213,8 +214,9 @@ class SubagentResult:
def update_bash_executions(self, executions: list[dict[str, Any]] | None) -> None:
"""Merge bash evidence from the latest yielded state while still running.
Entries merge by ``tool_call_id`` in first-seen order and are capped to
the newest ``_BASH_EVIDENCE_MAX_ENTRIES`` — unlike a terminal
Entries merge by their result message (``tool_call_id`` when it has no
id, since providers can reuse call ids across responses) in first-seen
order and are capped to the newest ``_BASH_EVIDENCE_MAX_ENTRIES`` — unlike a terminal
``final_state`` scan, accumulation survives summarization compacting
earlier AI/ToolMessages out of the streamed history. ``None`` leaves
the field untouched (no evidence this chunk); an empty list still
@ -227,9 +229,9 @@ class SubagentResult:
with self._state_lock:
if self.status.is_terminal:
return
merged = {str(entry.get("tool_call_id")): entry for entry in (self.bash_executions or [])}
merged = {_bash_evidence_key(entry): entry for entry in (self.bash_executions or [])}
for execution in executions:
merged[str(execution.get("tool_call_id"))] = dict(execution)
merged[_bash_evidence_key(execution)] = dict(execution)
self.bash_executions = list(merged.values())[-_BASH_EVIDENCE_MAX_ENTRIES:]
def snapshot_tool_receipts(self) -> list[dict[str, Any]] | None:
@ -531,10 +533,22 @@ def _harvest_shell_persistence(final_state: Any) -> bool | None:
return None
def _bash_evidence_key(execution: dict[str, Any]) -> tuple[str, str]:
"""Identify one recorded execution across values chunks.
A provider may reuse a call id in a later response, so the result message
id names the execution; entries without one keep the call-id identity.
"""
result_id = execution.get("result_id")
if isinstance(result_id, str) and result_id:
return ("result", result_id)
return ("call", str(execution.get("tool_call_id")))
def _harvest_bash_executions(
final_state: Any,
*,
summary_cache: dict[str, tuple[str, str]] | None = None,
summary_cache: dict[tuple[str, str], tuple[str, str]] | None = None,
) -> list[dict[str, Any]] | None:
"""Harvest bounded bash command/output evidence from one streamed state.
@ -564,48 +578,47 @@ def _harvest_bash_executions(
summary_cache.clear()
return None
try:
from deerflow.agents.middlewares.tool_call_args import pair_tool_call_results
from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY
from deerflow.subagents.acceptance_checks import summarize_test_output
commands: dict[str, tuple[str, str, bool]] = {}
for message in messages:
if not isinstance(message, AIMessage):
continue
for tool_call in message.tool_calls or []:
name = str(tool_call.get("name") or "")
if name not in _BASH_EVIDENCE_TOOL_NAMES:
continue
tool_call_id = str(tool_call.get("id") or "")
args = tool_call.get("args")
command = args.get("command") if isinstance(args, dict) else None
command = command if isinstance(command, str) else ""
# A truncated command loses its suffix; the matcher must not
# treat shell-structure analysis of the prefix as proof (a
# selection-changing suffix like ``-k smoke`` could be cut).
commands[tool_call_id] = (name, command[:_BASH_EVIDENCE_COMMAND_CHARS], len(command) > _BASH_EVIDENCE_COMMAND_CHARS)
if not commands:
# Tool-call ids are unique per response, not per conversation: pair each
# result with the call of its own response, so a reused id never labels
# one command's output with another command.
bash_calls = [occurrence for occurrence in pair_tool_call_results(messages) if occurrence.name in _BASH_EVIDENCE_TOOL_NAMES]
if not bash_calls:
if summary_cache is not None:
summary_cache.clear()
return []
positions = {id(message): index for index, message in enumerate(messages)}
answered = sorted((occurrence for occurrence in bash_calls if occurrence.result is not None), key=lambda occurrence: positions[id(occurrence.result)])
executions: list[dict[str, Any]] = []
# Only retained executions need full-output parsing. The per-run cache
# survives values chunks, but keeps at most the same newest 20 outputs.
tool_messages = [message for message in messages if isinstance(message, ToolMessage) and str(message.tool_call_id or "") in commands][-_BASH_EVIDENCE_MAX_ENTRIES:]
next_cache: dict[str, tuple[str, str]] = {}
for message in tool_messages:
tool_call_id = str(message.tool_call_id or "")
entry = commands[tool_call_id]
name, command, command_truncated = entry
next_cache: dict[tuple[str, str], tuple[str, str]] = {}
for occurrence in answered[-_BASH_EVIDENCE_MAX_ENTRIES:]:
message = occurrence.result
tool_call_id = occurrence.call_id
name = occurrence.name
command = occurrence.args.get("command")
command = command if isinstance(command, str) else ""
# A truncated command loses its suffix; the matcher must not
# treat shell-structure analysis of the prefix as proof (a
# selection-changing suffix like ``-k smoke`` could be cut).
command_truncated = len(command) > _BASH_EVIDENCE_COMMAND_CHARS
command = command[:_BASH_EVIDENCE_COMMAND_CHARS]
meta = (message.additional_kwargs or {}).get(TOOL_META_KEY) or {}
meta_status = str(meta.get("status") or getattr(message, "status", "success") or "success")
content = message.content if isinstance(message.content, str) else json.dumps(message.content, sort_keys=True, default=str)
status, status_marker = _bash_evidence_status(content, meta_status)
cached = summary_cache.get(tool_call_id) if summary_cache is not None else None
result_id = message.id if isinstance(message.id, str) and message.id else None
cache_key = _bash_evidence_key({"result_id": result_id, "tool_call_id": tool_call_id})
cached = summary_cache.get(cache_key) if summary_cache is not None else None
if cached is not None and (cached[0] is content or cached[0] == content):
test_summary = cached[1]
else:
test_summary = summarize_test_output(content)
next_cache[tool_call_id] = (content, test_summary)
next_cache[cache_key] = (content, test_summary)
output_tail = content[-_BASH_EVIDENCE_OUTPUT_TAIL_CHARS:]
if len(content) > _BASH_EVIDENCE_OUTPUT_TAIL_CHARS and content[-_BASH_EVIDENCE_OUTPUT_TAIL_CHARS - 1] != "\n":
# A clipped test name/log line can look like a summary after
@ -614,6 +627,7 @@ def _harvest_bash_executions(
executions.append(
{
"tool_call_id": tool_call_id,
"result_id": result_id,
"tool_name": name,
"command": command,
"command_truncated": command_truncated,
@ -1802,7 +1816,7 @@ class SubagentExecutor:
return None
return _harvest_tool_receipts(final_state, prefer_citing_turn=prefer_citing_turn)
bash_summary_cache: dict[str, tuple[str, str]] = {}
bash_summary_cache: dict[tuple[str, str], tuple[str, str]] = {}
def current_bash_executions() -> list[dict[str, Any]] | None:
# RFC #4651 PR4: evidence for tests_passed acceptance leaves.

175
backend/tests/test_acceptance_evidence_reused_call_ids.py Normal file
View File

@ -0,0 +1,175 @@
"""tests_passed evidence must follow each tool call, even when a provider reuses its id.
Tool-call ids are unique per assistant response, not per conversation: some
providers number calls per response, so two turns can both issue ``call_0``.
"""
import importlib.util
import sys
from pathlib import Path
import pytest
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import StructuredTool
from deerflow.config.app_config import AppConfig
from deerflow.subagents.acceptance_checks import check_acceptance_criteria
from deerflow.subagents.config import SubagentConfig
FAILED_RUN = "===== 1 failed, 2 passed in 0.10s =====\nExit Code: 1"
PASSED_RUN = "===== 3 passed in 0.10s ====="
@pytest.fixture
def executor_module(monkeypatch, tmp_path):
# conftest installs an executor stub. Load the production module separately
# so the real subagent graph streams, harvests and merges the evidence.
path = Path(__file__).parents[1] / "packages/harness/deerflow/subagents/executor.py"
spec = importlib.util.spec_from_file_location("_reused_call_id_executor", path)
module = importlib.util.module_from_spec(spec)
monkeypatch.setitem(sys.modules, spec.name, module)
spec.loader.exec_module(module)
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
# The scripted bash tool has no sandbox; declare fresh-shell provenance so
# the verdict depends only on which run the evidence attributes.
monkeypatch.setattr(module, "_harvest_shell_persistence", lambda state: False)
yield module
module._shutdown_isolated_subagent_loop()
class ScriptedModel(GenericFakeChatModel):
def bind_tools(self, tools, **kwargs):
return self
def _bash_call(call_id: str, command: str) -> AIMessage:
return AIMessage(content="", tool_calls=[{"name": "bash", "id": call_id, "args": {"command": command}}])
@pytest.mark.asyncio
async def test_latest_failing_run_under_a_reused_call_id_fails_tests_passed(executor_module, monkeypatch):
outputs = iter([FAILED_RUN, PASSED_RUN, FAILED_RUN])
commands = []
def bash(command: str) -> str:
"""Run a shell command."""
commands.append(command)
return next(outputs)
# The provider numbers calls per response: the third run reuses call_0.
responses = [
_bash_call("call_0", "pytest tests"),
_bash_call("call_1", "pytest tests -q"),
_bash_call("call_0", "pytest tests"),
AIMessage(content="Tests were run."),
]
model = ScriptedModel(messages=iter(responses))
monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: model)
app_config = AppConfig.model_validate(
{
"models": [{"name": "offline", "use": "langchain_openai:ChatOpenAI", "model": "offline"}],
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"summarization": {"enabled": False},
}
)
executor = executor_module.SubagentExecutor(
config=SubagentConfig(name="tester", description="Offline test runner", skills=[]),
tools=[StructuredTool.from_function(bash, name="bash")],
parent_model="offline",
app_config=app_config,
thread_id="reused-call-id-thread",
user_id="reused-call-id-user",
acceptance_criteria=["tests_passed:pytest tests"],
)
result = await executor._aexecute("Run the test suite")
assert result.status == executor_module.SubagentStatus.COMPLETED, result.error
assert commands == ["pytest tests", "pytest tests -q", "pytest tests"]
# Every run stays a separate entry, in execution order.
assert [(entry["tool_call_id"], entry["status"]) for entry in result.bash_executions] == [("call_0", "error"), ("call_1", "success"), ("call_0", "error")]
verdict = check_acceptance_criteria(["tests_passed:pytest tests"], bash_executions=result.bash_executions)
# The decisive latest run failed, so the leaf is a checked negative.
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (True, False), verdict
def test_reused_call_id_never_takes_another_commands_output(executor_module):
# A cancelled or timed-out subagent leaves its last call unanswered. Its
# reused id must not relabel the earlier subset run as the full suite.
messages = [
HumanMessage(content="Run the suite"),
_bash_call("call_0", "pytest tests/unit"),
ToolMessage(content=PASSED_RUN, tool_call_id="call_0"),
_bash_call("call_0", "pytest tests"),
]
executions = executor_module._harvest_bash_executions({"messages": messages})
assert [entry["command"] for entry in executions] == ["pytest tests/unit"]
verdict = check_acceptance_criteria(["tests_passed:pytest tests"], bash_executions=executions)
assert verdict["all_hold"] is False, verdict
def test_merge_keeps_each_occurrence_of_a_reused_call_id(executor_module):
first = executor_module._harvest_bash_executions({"messages": [_bash_call("call_0", "pytest tests"), ToolMessage(content=FAILED_RUN, tool_call_id="call_0", id="result-1")]})
later = executor_module._harvest_bash_executions(
{
"messages": [
_bash_call("call_0", "pytest tests"),
ToolMessage(content=FAILED_RUN, tool_call_id="call_0", id="result-1"),
_bash_call("call_1", "pytest tests -q"),
ToolMessage(content=PASSED_RUN, tool_call_id="call_1", id="result-2"),
_bash_call("call_0", "pytest tests"),
ToolMessage(content=FAILED_RUN, tool_call_id="call_0", id="result-3"),
]
}
)
result = executor_module.SubagentResult(task_id="task", trace_id="trace", status=executor_module.SubagentStatus.RUNNING)
result.update_bash_executions(first)
result.update_bash_executions(later)
# The reused id's later run is appended, not written back over the first.
assert [(entry["tool_call_id"], entry["status"]) for entry in result.bash_executions] == [("call_0", "error"), ("call_1", "success"), ("call_0", "error")]
verdict = check_acceptance_criteria(["tests_passed:pytest tests"], bash_executions=result.bash_executions)
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (True, False), verdict
def test_summary_cache_keeps_one_slot_per_run_of_a_reused_call_id(executor_module, monkeypatch):
from unittest.mock import Mock
from deerflow.subagents import acceptance_checks
classify = Mock(wraps=acceptance_checks.summarize_test_output)
monkeypatch.setattr(acceptance_checks, "summarize_test_output", classify)
messages = [
_bash_call("call_0", "pytest tests"),
ToolMessage(content=FAILED_RUN, tool_call_id="call_0", id="result-1"),
_bash_call("call_0", "pytest tests"),
ToolMessage(content=PASSED_RUN, tool_call_id="call_0", id="result-2"),
]
cache = {}
for _ in range(3):
executor_module._harvest_bash_executions({"messages": messages}, summary_cache=cache)
# Each run is classified once; values chunks reuse both cached summaries.
assert classify.call_count == 2
assert list(cache) == [("result", "result-1"), ("result", "result-2")]
@pytest.mark.parametrize("call_id", [None, ""])
def test_a_call_without_an_id_records_no_evidence(executor_module, call_id):
# Pairing needs a call id. Without one the owner of a result is unknown,
# so the harvest records nothing and the leaf stays UNVERIFIED.
messages = [
AIMessage(content="", tool_calls=[{"name": "bash", "id": call_id, "args": {"command": "pytest tests"}}]),
ToolMessage(content=PASSED_RUN, tool_call_id=call_id or ""),
]
executions = executor_module._harvest_bash_executions({"messages": messages})
assert executions == []
verdict = check_acceptance_criteria(["tests_passed:pytest tests"], bash_executions=executions)
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (False, False), verdict

2
backend/tests/test_node_test_acceptance.py
View File

@ -355,7 +355,7 @@ def test_summary_cache_is_bounded_and_tracks_compaction(executor_module, monkeyp
assert len(entries) == len(cache) == 20
assert classify.call_count == 20
executor_module._harvest_bash_executions({"messages": [ai, tools[-1]]}, summary_cache=cache)
assert list(cache) == ["call-24"]
assert list(cache) == [("call", "call-24")]
assert classify.call_count == 20
executor_module._harvest_bash_executions({"messages": [AIMessage(content="compacted")]}, summary_cache=cache)
assert cache == {}