github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-20 21:18:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: emit structured runtime metadata (follow-up#3887) (#3906)

Browse Source 
* feat: emit structured runtime metadata

* fix: avoid subagent import cycle in replay gateway

* fix: preserve legacy subtask result parsing

* refactor: tighten runtime metadata contracts

* fix(middleware): keep recovery hint on task exception wrapper content

The structured-metadata stamp overwrote the wrapper text with the bare
task-failure message, dropping the model-facing 'Continue with available
context, or choose an alternative tool.' guidance that every other tool
exception keeps. Append the shared hint after the formatted message.

* fix(subagents): require lowercase hex for result_sha256 reader

Length-only validation accepted any 64-char string; a faulty serializer
or relaying wrapper could store a non-digest value in the delegation
ledger. Enforce the producer's hexdigest shape with a fullmatch.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
AochenShen99 2026-07-04 11:27:19 +08:00 committed by GitHub
parent 9a088805d7
commit 66b9e7f212
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
34 changed files with 1496 additions and 677 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
backend/AGENTS.md
View File

@ -181,6 +181,12 @@ from deerflow.config import get_app_config
# from app.gateway.routers.uploads import ... # ← will fail CI
```
Package import hygiene: the `deerflow.agents` and `deerflow.subagents` package
roots expose heavyweight graph/executor entrypoints lazily. Internal modules
that only need lightweight types, config, or registries should import the
concrete submodule instead of adding eager package-root imports that pull in the
tool graph or subagent executor during state/schema imports.
### Agent System
**Lead Agent** (`packages/harness/deerflow/agents/lead_agent/agent.py`):
@ -215,7 +221,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the
8. **GuardrailMiddleware** - *(optional, if `guardrails.enabled`)* Pre-tool-call authorization via pluggable `GuardrailProvider`; returns an error ToolMessage on deny. Providers: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md)
9. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution
10. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Version gate on file writes (issue #3857): `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Marks live on messages, so summarization dropping the read result invalidates the gate automatically; writes never refresh marks, forcing a re-read between consecutive edits. Gate check + tool execution are serialized per (thread, path) so same-turn parallel writes cannot reuse one stale mark; on sandboxes whose `read_file` reports failures as `"Error: ..."` strings instead of raising (AIO/E2B), uninspectable targets fail open (creation proceeds, no mark stamped)
11. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting
11. **ToolErrorHandlingMiddleware** - Receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string.
**Lead-only middlewares** (`build_middlewares`, appended after the base):

4
backend/docs/summarization.md
View File

@ -2,6 +2,8 @@
DeerFlow includes automatic conversation summarization to handle long conversations that approach model token limits. When enabled, the system automatically condenses older messages while preserving recent context.
New checkpoints no longer use raw task-result or skill-read transcript content to derive durable context. The capture path consumes bounded structured metadata stamped on the corresponding `ToolMessage.additional_kwargs`; transcript text remains display/model content, not the state-capture protocol.
## Overview
The summarization feature uses LangChain's `SummarizationMiddleware` to monitor conversation history and trigger summarization based on configurable thresholds. When activated, it:
@ -161,7 +163,7 @@ The default LangChain prompt instructs the model to:
- Recent messages are preserved
- The generated prose summary is stored in `summary_text`
6. **AI/Tool Pair Protection**: The system ensures AI messages and their corresponding tool messages stay together
7. **Skill context channel**: Skill files read during the conversation (tool calls whose name is in `skill_file_read_tool_names` and whose path is under `skills.container_path`, narrowed to `.../SKILL.md`) are captured by `DurableContextMiddleware` into the checkpointed `skill_context` channel as references: `name`, `path`, a one-line `description` parsed in-memory from the file's frontmatter, and `loaded_at`, deduped by path. On every model call they are rendered into a hidden durable-context data message as a compact "active skills" reminder that points at each `SKILL.md` for on-demand re-read, so which skills are active survives summarization without persisting or re-injecting the verbatim body. The channel keeps the most recently read skills (cap `_SKILL_CONTEXT_MAX_ENTRIES`; re-reading an existing skill refreshes its recency); sessions typically load only 1-3.
7. **Skill context channel**: Skill files read during the conversation (tool calls whose name is in `skill_file_read_tool_names` and whose path is under `skills.container_path`, narrowed to `.../SKILL.md`) are stamped with `skill_context_entry` metadata at the read-tool boundary, then captured by `DurableContextMiddleware` into the checkpointed `skill_context` channel as references: `name`, `path`, a one-line `description` parsed in-memory from the file's frontmatter, and `loaded_at`, deduped by path. On every model call they are rendered into a hidden durable-context data message as a compact "active skills" reminder that points at each `SKILL.md` for on-demand re-read, so which skills are active survives summarization without persisting or re-injecting the verbatim body. The channel keeps the most recently read skills (cap `_SKILL_CONTEXT_MAX_ENTRIES`; re-reading an existing skill refreshes its recency); sessions typically load only 1-3.
### Token Counting

35
backend/packages/harness/deerflow/agents/__init__.py
View File

@ -1,13 +1,4 @@
from .factory import create_deerflow_agent
from .features import Next, Prev, RuntimeFeatures
from .lead_agent import make_lead_agent
from .lead_agent.prompt import prime_enabled_skills_cache
from .thread_state import SandboxState, ThreadState
# LangGraph imports deerflow.agents when registering the graph. Prime the
# enabled-skills cache here so the request path can usually read a warm cache
# without forcing synchronous filesystem work during prompt module import.
prime_enabled_skills_cache()
__all__ = [
"create_deerflow_agent",
@ -18,3 +9,29 @@ __all__ = [
"SandboxState",
"ThreadState",
]
def __getattr__(name: str):
if name == "create_deerflow_agent":
from .factory import create_deerflow_agent
globals()[name] = create_deerflow_agent
return create_deerflow_agent
if name == "make_lead_agent":
from .lead_agent import make_lead_agent
from .lead_agent.prompt import prime_enabled_skills_cache
# LangGraph resolves deerflow.agents:make_lead_agent when registering
# the graph. Prime at that explicit entrypoint instead of at package
# import time so lightweight submodules can be imported without pulling
# in the whole tool/subagent graph.
prime_enabled_skills_cache()
globals()[name] = make_lead_agent
return make_lead_agent
if name in {"SandboxState", "ThreadState"}:
from .thread_state import SandboxState, ThreadState
exports = {"SandboxState": SandboxState, "ThreadState": ThreadState}
globals().update(exports)
return exports[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

3
backend/packages/harness/deerflow/agents/lead_agent/prompt.py
View File

@ -7,6 +7,7 @@ from functools import lru_cache
from typing import TYPE_CHECKING
from deerflow.config.agents_config import load_agent_soul
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.types import Skill, SkillCategory
from deerflow.subagents import get_available_subagent_names
@ -675,7 +676,7 @@ def get_skills_prompt_section(available_skills: set[str] | None = None, *, app_c
container_base_path = config.skills.container_path
skill_evolution_enabled = config.skill_evolution.enabled
except Exception:
container_base_path = "/mnt/skills"
container_base_path = DEFAULT_SKILLS_CONTAINER_PATH
skill_evolution_enabled = False
else:
config = app_config

54
backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py
View File

@ -10,15 +10,20 @@ from typing import Any
from langchain_core.messages import AIMessage, AnyMessage, ToolMessage
from deerflow.agents.thread_state import DelegationEntry
from deerflow.subagents.status_contract import SUBAGENT_STATUS_KEY, extract_subagent_status
from deerflow.subagents.status_contract import (
read_subagent_result_metadata,
)
_RESULT_BRIEF_CAP = 2000
_DESCRIPTION_CAP = 200
_LEDGER_RENDER_CHAR_BUDGET = 6000
_LEDGER_ENTRY_RESULT_RENDER_CAP = 120
_TASK_SUCCESS_PREFIX = "Task Succeeded. Result:"
_TASK_FAILED_PREFIX = "Task failed. Error:"
_TASK_TIMED_OUT_PREFIX = "Task timed out. Error:"
_STATUS_ONLY_RESULT_BRIEFS = {
"failed": "Task failed.",
"cancelled": "Task cancelled by user.",
"timed_out": "Task timed out.",
"polling_timed_out": "Task polling timed out.",
}
def _utc_now_iso() -> str:
@ -41,20 +46,6 @@ def _bound_text(text: str, cap: int = _RESULT_BRIEF_CAP) -> str:
return f"{text[:head]}{omitted_marker}{text[-tail:]}"
def _parse_task_result(content: str, status: str | None = None) -> tuple[str, str] | None:
text = (content if isinstance(content, str) else str(content)).strip()
status = status or extract_subagent_status(text)
if status is None:
return None
if status == "completed" and text.startswith(_TASK_SUCCESS_PREFIX):
return status, text[len(_TASK_SUCCESS_PREFIX) :].strip()
if status == "failed" and text.startswith(_TASK_FAILED_PREFIX):
return status, text[len(_TASK_FAILED_PREFIX) :].strip()
if status == "timed_out" and text.startswith(_TASK_TIMED_OUT_PREFIX):
return status, text[len(_TASK_TIMED_OUT_PREFIX) :].strip()
return status, text
def _escape_context_text(value: object) -> str:
return escape(" ".join(str(value).split()), quote=False)
@ -128,21 +119,20 @@ def extract_delegations(messages: list[AnyMessage]) -> list[DelegationEntry]:
entry = entries_by_id.get(tool_call_id)
if entry is None:
continue
content = message.content if isinstance(message.content, str) else str(message.content)
status = message.additional_kwargs.get(SUBAGENT_STATUS_KEY)
parsed = _parse_task_result(content, status if isinstance(status, str) else None)
if parsed is None:
structured = read_subagent_result_metadata(message.additional_kwargs)
if structured is None:
continue
status, result_text = parsed
result_ref = str(message.id or tool_call_id)
entry.update(
{
"status": status,
"result_brief": _bound_text(result_text),
"result_sha256": hashlib.sha256(result_text.encode("utf-8")).hexdigest(),
"result_ref": result_ref,
}
)
entry["status"] = structured["status"]
result_text = structured.get("result_brief") or structured.get("error") or _STATUS_ONLY_RESULT_BRIEFS.get(structured["status"])
if result_text:
result_sha256 = structured.get("result_sha256") or hashlib.sha256(result_text.encode("utf-8")).hexdigest()
entry.update(
{
"result_brief": _bound_text(result_text),
"result_sha256": result_sha256,
"result_ref": str(message.id or tool_call_id),
}
)
return [entries_by_id[tool_call_id] for tool_call_id in order]

8
backend/packages/harness/deerflow/agents/middlewares/durable_context_middleware.py
View File

@ -23,9 +23,9 @@ from langgraph.runtime import Runtime
from deerflow.agents.middlewares.delegation_ledger import extract_delegations, render_delegation_ledger
from deerflow.agents.middlewares.skill_context import extract_skills, render_skill_context
from deerflow.agents.thread_state import _DELEGATION_LEDGER_MAX_ENTRIES, TERMINAL_STATUSES
from deerflow.config.summarization_config import DEFAULT_SKILL_FILE_READ_TOOL_NAMES
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
_DEFAULT_SKILLS_ROOT = "/mnt/skills"
_DEFAULT_SKILL_READ_TOOL_NAMES = frozenset({"read_file", "read", "view", "cat"})
_DURABLE_CONTEXT_DATA_KEY = "durable_context_data"
_SUMMARY_RENDER_CHAR_BUDGET = 6000
_AUTHORITY_CONTRACT = "\n".join(
@ -40,7 +40,7 @@ _DELEGATION_STABLE_FIELDS = ("description", "subagent_type", "status", "result_b
def _normalize_skills_root(skills_container_path: str | None) -> str:
return posixpath.normpath(skills_container_path or _DEFAULT_SKILLS_ROOT)
return posixpath.normpath(skills_container_path or DEFAULT_SKILLS_CONTAINER_PATH)
def _bound_text(text: str, cap: int) -> str:
@ -124,7 +124,7 @@ class DurableContextMiddleware(AgentMiddleware[AgentState]):
) -> None:
super().__init__()
self._skills_root = _normalize_skills_root(skills_container_path)
self._skill_read_tool_names = frozenset(_DEFAULT_SKILL_READ_TOOL_NAMES if skill_file_read_tool_names is None else skill_file_read_tool_names)
self._skill_read_tool_names = frozenset(DEFAULT_SKILL_FILE_READ_TOOL_NAMES if skill_file_read_tool_names is None else skill_file_read_tool_names)
@override
def before_model(self, state: AgentState, runtime: Runtime) -> dict | None:

69
backend/packages/harness/deerflow/agents/middlewares/skill_context.py
View File

@ -2,11 +2,12 @@
from __future__ import annotations
import logging
import posixpath
import re
from collections.abc import Collection
from collections.abc import Collection, Mapping
from html import escape
from typing import Any
from typing import Any, TypedDict
import yaml
from langchain_core.messages import AIMessage, AnyMessage, ToolMessage
@ -15,6 +16,13 @@ from deerflow.agents.thread_state import _SKILL_DESCRIPTION_MAX_CHARS, SkillEntr
_SKILL_FILE_NAME = "SKILL.md"
_FRONT_MATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
SKILL_CONTEXT_ENTRY_KEY = "skill_context_entry"
logger = logging.getLogger(__name__)
class SkillEntryMetadata(TypedDict):
path: str
description: str
def _tool_call_name(tool_call: dict[str, Any]) -> str:
@ -80,6 +88,38 @@ def _is_tool_error_text(content: str) -> bool:
return content.lstrip().startswith("Error:")
def build_skill_entry_metadata_from_read(
path: str,
content: str,
*,
skills_root: str,
) -> SkillEntryMetadata | None:
normalized_root = posixpath.normpath(skills_root.rstrip("/") or "/")
normalized_path = _normalize_under_root(path, normalized_root)
if normalized_path is None or not _is_skill_file(normalized_path) or _is_tool_error_text(content):
return None
return {
"path": normalized_path,
"description": _parse_description(content),
}
def read_skill_entry_metadata(additional_kwargs: Mapping[str, object] | None) -> SkillEntryMetadata | None:
if not additional_kwargs:
return None
raw = additional_kwargs.get(SKILL_CONTEXT_ENTRY_KEY)
if not isinstance(raw, Mapping):
return None
path = raw.get("path")
description = raw.get("description")
if not isinstance(path, str):
return None
return {
"path": path,
"description": " ".join(description.split())[:_SKILL_DESCRIPTION_MAX_CHARS] if isinstance(description, str) else "",
}
def _escape_context_text(value: object) -> str:
return escape(str(value), quote=False)
@ -114,17 +154,28 @@ def extract_skills(
if getattr(message, "status", "success") == "error":
continue
tool_call_id = str(message.tool_call_id) if message.tool_call_id else ""
path = skill_paths_by_id.get(tool_call_id)
if path is None:
expected_path = skill_paths_by_id.get(tool_call_id)
if expected_path is None:
continue
content = message.content if isinstance(message.content, str) else str(message.content)
if _is_tool_error_text(content):
metadata = read_skill_entry_metadata(message.additional_kwargs)
if metadata is None:
content = message.content if isinstance(message.content, str) else ""
if not _is_tool_error_text(content):
logger.warning("missing skill read metadata: tool_call_id=%s path=%s", tool_call_id, expected_path)
continue
if metadata["path"] != expected_path:
logger.warning(
"mismatched skill read metadata: tool_call_id=%s expected_path=%s metadata_path=%s",
tool_call_id,
expected_path,
metadata["path"],
)
continue
entries.append(
{
"name": _skill_name_from_path(path),
"path": path,
"description": _parse_description(content),
"name": _skill_name_from_path(expected_path),
"path": expected_path,
"description": metadata["description"],
"loaded_at": index,
}
)

98
backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py
View File

@ -11,9 +11,16 @@ from langgraph.errors import GraphBubbleUp
from langgraph.prebuilt.tool_node import ToolCallRequest
from langgraph.types import Command
from deerflow.agents.middlewares.skill_context import (
SKILL_CONTEXT_ENTRY_KEY,
_tool_call_path,
build_skill_entry_metadata_from_read,
)
from deerflow.config.app_config import AppConfig
from deerflow.config.summarization_config import DEFAULT_SKILL_FILE_READ_TOOL_NAMES
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
from deerflow.subagents.status_contract import (
extract_subagent_status,
format_subagent_result_message,
make_subagent_additional_kwargs,
)
@ -24,34 +31,19 @@ logger = logging.getLogger(__name__)
_MISSING_TOOL_CALL_ID = "missing_tool_call_id"
_TASK_TOOL_NAME = "task"
_RECOVERY_HINT = "Continue with available context, or choose an alternative tool."
def _stamp_task_subagent_status(message: ToolMessage, *, tool_name: str, error: str | None = None) -> ToolMessage:
"""Centralised stamping of ``additional_kwargs.subagent_status``.
Bytedance/deer-flow issue #3146: the frontend now reads the subagent
status from a structured field instead of parsing the leading text of
the task tool's return string. That contract is enforced here, in the
one place every task tool result flows through, rather than at the 5
normal-return + 3 ``Error:`` pre-execution branches inside
``task_tool.py``. Centralisation prevents the "added a new return
path, forgot the stamp" drift mode.
For non-``task`` tools this is a no-op so other tools' additional_kwargs
conventions are untouched.
"""
def _stamp_task_exception_status(message: ToolMessage, *, tool_name: str, error: str) -> ToolMessage:
"""Stamp failed metadata on task exception wrappers produced here."""
if tool_name != _TASK_TOOL_NAME:
return message
content = message.content if isinstance(message.content, str) else ""
status = extract_subagent_status(content)
if status is None:
# Non-terminal streaming chunks or unrecognised shapes leave the
# field unset so the frontend can keep the card on its in-progress
# placeholder until a real terminal frame arrives.
return message
stamp = make_subagent_additional_kwargs(status, error=error)
content, metadata_error = format_subagent_result_message("failed", error=error)
if not content.endswith((".", "!", "?")):
content += "."
message.content = f"{content} {_RECOVERY_HINT}"
existing = dict(message.additional_kwargs or {})
existing.update(stamp)
existing.update(make_subagent_additional_kwargs("failed", error=metadata_error))
message.additional_kwargs = existing
return message
@ -59,6 +51,16 @@ def _stamp_task_subagent_status(message: ToolMessage, *, tool_name: str, error:
class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]):
"""Convert tool exceptions into error ToolMessages so the run can continue."""
def __init__(self, *, app_config: AppConfig | None = None) -> None:
super().__init__()
self._app_config = app_config
if app_config is None:
self._skill_read_tool_names = frozenset(DEFAULT_SKILL_FILE_READ_TOOL_NAMES)
self._skills_root = DEFAULT_SKILLS_CONTAINER_PATH
else:
self._skill_read_tool_names = frozenset(app_config.summarization.skill_file_read_tool_names)
self._skills_root = app_config.skills.container_path
def _build_error_message(self, request: ToolCallRequest, exc: Exception) -> ToolMessage:
tool_name = str(request.tool_call.get("name") or "unknown_tool")
tool_call_id = str(request.tool_call.get("id") or _MISSING_TOOL_CALL_ID)
@ -66,32 +68,50 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]):
if len(detail) > 500:
detail = detail[:497] + "..."
content = f"Error: Tool '{tool_name}' failed with {exc.__class__.__name__}: {detail}. Continue with available context, or choose an alternative tool."
content = f"Error: Tool '{tool_name}' failed with {exc.__class__.__name__}: {detail}. {_RECOVERY_HINT}"
message = ToolMessage(
content=content,
tool_call_id=tool_call_id,
name=tool_name,
status="error",
)
# Stamp the structured subagent status on the wrapper too: the
# frontend would otherwise have to fall back to prefix-matching
# ``Error: Tool 'task' failed ...`` on the wire. The ``subagent_error``
# carries the same ``ExcClass: detail`` shape the wrapper string
# uses so debugging artifacts stay aligned.
# This middleware is the producer for exception wrappers, so task
# failures raised before task_tool can build its own Command still
# carry the same structured metadata.
structured_error = f"{exc.__class__.__name__}: {detail}"
return _stamp_task_subagent_status(message, tool_name=tool_name, error=structured_error)
return _stamp_task_exception_status(message, tool_name=tool_name, error=structured_error)
@staticmethod
def _maybe_stamp(result: ToolMessage | Command, request: ToolCallRequest) -> ToolMessage | Command:
"""Apply the subagent stamp to successful task tool returns.
def _stamp_skill_read_metadata(
self,
message: ToolMessage,
request: ToolCallRequest,
*,
tool_name: str,
) -> ToolMessage:
if tool_name not in self._skill_read_tool_names:
return message
if getattr(message, "status", "success") == "error":
return message
content = message.content if isinstance(message.content, str) else None
if content is None:
return message
path = _tool_call_path(request.tool_call)
if path is None:
return message
entry = build_skill_entry_metadata_from_read(path, content, skills_root=self._skills_root)
if entry is None:
return message
existing = dict(message.additional_kwargs or {})
existing[SKILL_CONTEXT_ENTRY_KEY] = dict(entry)
message.additional_kwargs = existing
return message
``Command`` results bypass the stamp — they encode LangGraph
control flow rather than user-facing tool output.
"""
def _maybe_stamp(self, result: ToolMessage | Command, request: ToolCallRequest) -> ToolMessage | Command:
"""Apply producer-bound metadata for tool results that need it."""
if not isinstance(result, ToolMessage):
return result
tool_name = str(request.tool_call.get("name") or "")
return _stamp_task_subagent_status(result, tool_name=tool_name)
return self._stamp_skill_read_metadata(result, request, tool_name=tool_name)
@override
def wrap_tool_call(
@ -198,7 +218,7 @@ def _build_runtime_middlewares(
tail.append(ReadBeforeWriteMiddleware())
tail.append(ToolErrorHandlingMiddleware())
tail.append(ToolErrorHandlingMiddleware(app_config=app_config))
return [*outer_wrappers, *thread_hooks, *tail]

3
backend/packages/harness/deerflow/config/skills_config.py
View File

@ -4,6 +4,7 @@ from pathlib import Path
from pydantic import BaseModel, Field
from deerflow.config.runtime_paths import project_root, resolve_path
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
def _legacy_skills_candidates() -> tuple[Path, ...]:
@ -25,7 +26,7 @@ class SkillsConfig(BaseModel):
description=("Path to skills directory. If not specified, defaults to `skills` under the caller project root, falling back to the legacy repo-root location for monorepo compatibility."),
)
container_path: str = Field(
default="/mnt/skills",
default=DEFAULT_SKILLS_CONTAINER_PATH,
description="Path where skills are mounted in the sandbox container",
)

3
backend/packages/harness/deerflow/config/summarization_config.py
View File

@ -5,6 +5,7 @@ from typing import Literal
from pydantic import BaseModel, Field
ContextSizeType = Literal["fraction", "tokens", "messages"]
DEFAULT_SKILL_FILE_READ_TOOL_NAMES: tuple[str, ...] = ("read_file", "read", "view", "cat")
class ContextSize(BaseModel):
@ -52,7 +53,7 @@ class SummarizationConfig(BaseModel):
description="Custom prompt template for generating summaries. If not provided, uses the default LangChain prompt.",
)
skill_file_read_tool_names: list[str] = Field(
default_factory=lambda: ["read_file", "read", "view", "cat"],
default_factory=lambda: list(DEFAULT_SKILL_FILE_READ_TOOL_NAMES),
description="Tool names treated as skill-file reads when capturing loaded skills into the durable skill_context channel.",
)

3
backend/packages/harness/deerflow/constants.py Normal file
View File

@ -0,0 +1,3 @@
"""Shared runtime protocol constants."""
DEFAULT_SKILLS_CONTAINER_PATH = "/mnt/skills"

3
backend/packages/harness/deerflow/sandbox/tools.py
View File

@ -12,6 +12,7 @@ from langchain.tools import tool
from deerflow.agents.thread_state import ThreadDataState
from deerflow.config import get_app_config
from deerflow.config.paths import VIRTUAL_PATH_PREFIX
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
from deerflow.runtime.secret_context import read_active_secrets
from deerflow.runtime.user_context import resolve_runtime_user_id
from deerflow.sandbox.exceptions import (
@ -45,7 +46,7 @@ _LOCAL_BASH_SYSTEM_PATH_PREFIXES = (
"/dev/",
)
_DEFAULT_SKILLS_CONTAINER_PATH = "/mnt/skills"
_DEFAULT_SKILLS_CONTAINER_PATH = DEFAULT_SKILLS_CONTAINER_PATH
_ACP_WORKSPACE_VIRTUAL_PATH = "/mnt/acp-workspace"
_DEFAULT_GLOB_MAX_RESULTS = 200
_MAX_GLOB_MAX_RESULTS = 1000

3
backend/packages/harness/deerflow/skills/slash.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import re
from dataclasses import dataclass
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
from deerflow.skills.types import Skill
RESERVED_SLASH_SKILL_NAMES = frozenset({"bootstrap", "goal", "help", "memory", "models", "new", "status"})
@ -45,7 +46,7 @@ def resolve_slash_skill(
skills: list[Skill],
*,
available_skills: set[str] | None = None,
container_base_path: str = "/mnt/skills",
container_base_path: str = DEFAULT_SKILLS_CONTAINER_PATH,
) -> ResolvedSlashSkill | None:
"""Resolve text into an enabled, whitelisted skill activation if possible."""
reference = parse_slash_skill_reference(text)

3
backend/packages/harness/deerflow/skills/storage/local_skill_storage.py
View File

@ -14,14 +14,13 @@ from datetime import UTC, datetime
from pathlib import Path
from deerflow.config.runtime_paths import resolve_path
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
from deerflow.skills.permissions import make_skill_written_path_sandbox_readable
from deerflow.skills.storage.skill_storage import SKILL_MD_FILE, SkillStorage
from deerflow.skills.types import SkillCategory
logger = logging.getLogger(__name__)
DEFAULT_SKILLS_CONTAINER_PATH = "/mnt/skills"
# Bound for the best-effort temp-dir cleanup so a stalled filesystem (e.g. NFS)
# cannot hold back the install outcome propagating out of the finally block.
_INSTALL_TMP_CLEANUP_TIMEOUT_SECONDS = 5.0

3
backend/packages/harness/deerflow/skills/storage/skill_storage.py
View File

@ -8,6 +8,7 @@ from abc import ABC, abstractmethod
from collections.abc import Iterable
from pathlib import Path
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
from deerflow.skills.types import SKILL_MD_FILE, Skill, SkillCategory # noqa: F401
logger = logging.getLogger(__name__)
@ -24,7 +25,7 @@ class SkillStorage(ABC):
compose them with protocol-level helpers.
"""
def __init__(self, container_path: str = "/mnt/skills") -> None:
def __init__(self, container_path: str = DEFAULT_SKILLS_CONTAINER_PATH) -> None:
self._container_root = container_path
# ------------------------------------------------------------------

6
backend/packages/harness/deerflow/skills/types.py
View File

@ -2,6 +2,8 @@ from dataclasses import dataclass, field
from enum import StrEnum
from pathlib import Path
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
SKILL_MD_FILE = "SKILL.md"
@ -50,7 +52,7 @@ class Skill:
path = self.relative_path.as_posix()
return "" if path == "." else path
def get_container_path(self, container_base_path: str = "/mnt/skills") -> str:
def get_container_path(self, container_base_path: str = DEFAULT_SKILLS_CONTAINER_PATH) -> str:
"""
Get the full path to this skill in the container.
@ -66,7 +68,7 @@ class Skill:
return f"{category_base}/{skill_path}"
return category_base
def get_container_file_path(self, container_base_path: str = "/mnt/skills") -> str:
def get_container_file_path(self, container_base_path: str = DEFAULT_SKILLS_CONTAINER_PATH) -> str:
"""
Get the full path to this skill's main file (SKILL.md) in the container.

14
backend/packages/harness/deerflow/subagents/__init__.py
View File

@ -1,5 +1,4 @@
from .config import SubagentConfig
from .executor import SubagentExecutor, SubagentResult
from .registry import get_available_subagent_names, get_subagent_config, list_subagents
__all__ = [
@ -10,3 +9,16 @@ __all__ = [
"get_subagent_config",
"list_subagents",
]
def __getattr__(name: str):
if name in {"SubagentExecutor", "SubagentResult"}:
from .executor import SubagentExecutor, SubagentResult
exports = {
"SubagentExecutor": SubagentExecutor,
"SubagentResult": SubagentResult,
}
globals().update(exports)
return exports[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

150
backend/packages/harness/deerflow/subagents/status_contract.py
View File

@ -1,31 +1,36 @@
"""Backend↔frontend contract for the structured subagent status.
"""Backend↔frontend contract for structured subagent result metadata.
Bytedance/deer-flow issue #3146: the frontend used to derive the
subtask card state by string-matching the leading text of the
``task`` tool's result. That contract was fragile — any rewording on
the backend silently broke the card lifecycle, and the issue history
of #3107 BUG-007 / #3131 review showed it repeatedly.
This module replaces the text-shaped contract with a small structured
one carried inside ``ToolMessage.additional_kwargs``:
``task`` tool result text is model-visible display content. Runtime
consumers read the structured facts carried inside
``ToolMessage.additional_kwargs``:
- ``subagent_status``: one of ``SUBAGENT_STATUS_VALUES``.
- ``subagent_error`` (optional): the human-readable error blob the
backend recorded.
- ``subagent_result_brief`` / ``subagent_result_sha256`` (optional):
bounded completed-result metadata plus a digest of the full result.
The mapping from "task tool result text" to status is the one piece
the backend stamper (``ToolErrorHandlingMiddleware``) and the
frontend fallback parser must agree on. The shared fixture at
``contracts/subagent_status_contract.json`` is the single source of
truth — both sides' tests load it and assert behaviour.
The shared fixture at ``contracts/subagent_status_contract.json`` pins
the enum values across Python and TypeScript.
"""
from __future__ import annotations
from typing import Literal
import hashlib
import re
from collections.abc import Mapping
from typing import Literal, NotRequired, TypedDict
SUBAGENT_STATUS_KEY = "subagent_status"
SUBAGENT_ERROR_KEY = "subagent_error"
SUBAGENT_RESULT_BRIEF_KEY = "subagent_result_brief"
SUBAGENT_RESULT_SHA256_KEY = "subagent_result_sha256"
SUBAGENT_METADATA_TEXT_MAX_CHARS = 2000
#: The producer always emits ``hashlib.sha256(...).hexdigest()`` — 64
#: lowercase hex chars. Readers enforce the same shape so a corrupted
#: relay value cannot masquerade as a digest.
_SHA256_HEX_RE = re.compile(r"[0-9a-f]{64}")
SubagentStatusValue = Literal[
"completed",
@ -46,41 +51,32 @@ SUBAGENT_STATUS_VALUES: tuple[SubagentStatusValue, ...] = (
"polling_timed_out",
)
# Prefix table — ordered most-specific-first because some prefixes are
# substrings of others ("Task timed out" vs "Task polling timed out", "Task
# failed" vs "Task failed. Error: ..."). The "Task " prefixes come from
# ``task_tool.py``'s 5 normal-return strings; the bare ``Error:`` prefix
# catches both the 3 ``Error:`` pre-execution returns and the wrapper
# produced by ``ToolErrorHandlingMiddleware`` for any task tool exception.
_PREFIX_TO_STATUS: tuple[tuple[str, SubagentStatusValue], ...] = (
("Task Succeeded. Result:", "completed"),
("Task polling timed out", "polling_timed_out"),
("Task timed out", "timed_out"),
("Task cancelled by user", "cancelled"),
("Task failed.", "failed"),
("Error", "failed"),
)
class StructuredSubagentResult(TypedDict):
status: SubagentStatusValue
result_brief: NotRequired[str]
result_sha256: NotRequired[str]
error: NotRequired[str]
def extract_subagent_status(content: str) -> SubagentStatusValue | None:
"""Infer the structured status for a ``task`` tool result string.
Returns ``None`` when the content does not match any known terminal
prefix. Non-terminal streaming chunks fall into this branch by
design — the middleware then leaves ``subagent_status`` unset so
the frontend keeps the card on its in-progress placeholder until
the real terminal frame arrives.
"""
trimmed = content.strip()
for prefix, status in _PREFIX_TO_STATUS:
if trimmed.startswith(prefix):
return status
return None
def _bound_metadata_text(text: str, cap: int = SUBAGENT_METADATA_TEXT_MAX_CHARS) -> str:
cleaned = text.strip()
if len(cleaned) <= cap:
return cleaned
marker = "\n...\n"
if cap <= len(marker):
return cleaned[:cap]
head = cap * 2 // 3
tail = cap - head - len(marker)
if tail <= 0:
return cleaned[:cap]
return f"{cleaned[:head]}{marker}{cleaned[-tail:]}"
def make_subagent_additional_kwargs(
status: SubagentStatusValue,
*,
result: str | None = None,
error: str | None = None,
) -> dict[str, str]:
"""Build the ``additional_kwargs`` payload the middleware stamps.
@ -91,12 +87,72 @@ def make_subagent_additional_kwargs(
Raises:
ValueError: when ``status`` is not in :data:`SUBAGENT_STATUS_VALUES`.
We do not accept arbitrary strings: a typo would silently leak
through to the frontend and degrade to the legacy prefix
fallback rather than failing loudly.
through to consumers as missing metadata rather than failing
loudly at the producer boundary.
"""
if status not in SUBAGENT_STATUS_VALUES:
raise ValueError(f"invalid subagent status {status!r}; expected one of {SUBAGENT_STATUS_VALUES}")
payload: dict[str, str] = {SUBAGENT_STATUS_KEY: status}
if error and error.strip():
payload[SUBAGENT_ERROR_KEY] = error.strip()
if status == "completed" and isinstance(result, str) and result.strip():
payload[SUBAGENT_RESULT_BRIEF_KEY] = _bound_metadata_text(result)
payload[SUBAGENT_RESULT_SHA256_KEY] = hashlib.sha256(result.encode("utf-8")).hexdigest()
if status != "completed" and isinstance(error, str) and error.strip():
payload[SUBAGENT_ERROR_KEY] = _bound_metadata_text(error)
return payload
def format_subagent_result_message(
status: SubagentStatusValue,
*,
result: str | None = None,
error: str | None = None,
) -> tuple[str, str | None]:
"""Return model-visible task content plus normalized metadata error."""
result_text = "" if result is None else str(result)
error_text = str(error).strip() if isinstance(error, str) else ""
if status == "completed":
return f"Task Succeeded. Result: {result_text}", None
if status == "cancelled":
detail = error_text or "Task cancelled by user."
if detail == "Task cancelled by user.":
return detail, detail
return f"Task cancelled by user. Error: {detail}", detail
if status == "timed_out":
detail = error_text or "Task timed out."
if detail == "Task timed out.":
return detail, detail
return f"Task timed out. Error: {detail}", detail
if status == "polling_timed_out":
detail = error_text or "Task polling timed out."
return detail, detail
detail = error_text or "Task failed."
if detail == "Task failed.":
return detail, detail
return f"Task failed. Error: {detail}", detail
def read_subagent_result_metadata(
additional_kwargs: Mapping[str, object] | None,
) -> StructuredSubagentResult | None:
if not additional_kwargs:
return None
raw_status = additional_kwargs.get(SUBAGENT_STATUS_KEY)
if raw_status not in SUBAGENT_STATUS_VALUES:
return None
status = raw_status
payload: StructuredSubagentResult = {"status": status}
raw_result = additional_kwargs.get(SUBAGENT_RESULT_BRIEF_KEY)
raw_hash = additional_kwargs.get(SUBAGENT_RESULT_SHA256_KEY)
raw_error = additional_kwargs.get(SUBAGENT_ERROR_KEY)
if status == "completed" and isinstance(raw_result, str) and raw_result.strip():
payload["result_brief"] = _bound_metadata_text(raw_result)
if isinstance(raw_hash, str) and _SHA256_HEX_RE.fullmatch(raw_hash):
payload["result_sha256"] = raw_hash
if status != "completed" and isinstance(raw_error, str) and raw_error.strip():
payload["error"] = _bound_metadata_text(raw_error)
return payload

82
backend/packages/harness/deerflow/tools/builtins/task_tool.py
View File

@ -8,7 +8,9 @@ from typing import TYPE_CHECKING, Annotated, Any, cast
from langchain.tools import InjectedToolCallId, tool
from langchain_core.callbacks import BaseCallbackManager
from langchain_core.messages import ToolMessage
from langgraph.config import get_stream_writer
from langgraph.types import Command
from deerflow.config import get_app_config
from deerflow.runtime.user_context import resolve_runtime_user_id
@ -21,6 +23,11 @@ from deerflow.subagents.executor import (
get_background_task_result,
request_cancel_background_task,
)
from deerflow.subagents.status_contract import (
SubagentStatusValue,
format_subagent_result_message,
make_subagent_additional_kwargs,
)
from deerflow.tools.types import Runtime
from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id
@ -185,6 +192,28 @@ def _merge_skill_allowlists(parent: list[str] | None, child: list[str] | None) -
return [skill for skill in child if skill in parent_set]
def _task_result_command(
*,
tool_call_id: str,
status: SubagentStatusValue,
result: str | None = None,
error: str | None = None,
) -> Command:
content, metadata_error = format_subagent_result_message(status, result=result, error=error)
return Command(
update={
"messages": [
ToolMessage(
content=content,
tool_call_id=tool_call_id,
name="task",
additional_kwargs=make_subagent_additional_kwargs(status, result=result, error=metadata_error),
)
]
}
)
@tool("task", parse_docstring=True)
async def task_tool(
runtime: Runtime,
@ -192,7 +221,7 @@ async def task_tool(
prompt: str,
subagent_type: str,
tool_call_id: Annotated[str, InjectedToolCallId],
) -> str:
) -> str | Command:
"""Delegate a task to a specialized subagent that runs in its own context.
Subagents help you:
@ -236,11 +265,20 @@ async def task_tool(
config = get_subagent_config(subagent_type, app_config=runtime_app_config) if runtime_app_config is not None else get_subagent_config(subagent_type)
if config is None:
available = ", ".join(available_subagent_names)
return f"Error: Unknown subagent type '{subagent_type}'. Available: {available}"
error = f"Unknown subagent type '{subagent_type}'. Available: {available}"
return _task_result_command(
tool_call_id=tool_call_id,
status="failed",
error=error,
)
if subagent_type == "bash":
host_bash_allowed = is_host_bash_allowed(runtime_app_config) if runtime_app_config is not None else is_host_bash_allowed()
if not host_bash_allowed:
return f"Error: {LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE}"
return _task_result_command(
tool_call_id=tool_call_id,
status="failed",
error=LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE,
)
# Build config overrides
overrides: dict = {}
@ -364,7 +402,12 @@ async def task_tool(
logger.error(f"[trace={trace_id}] Task {task_id} not found in background tasks")
writer({"type": "task_failed", "task_id": task_id, "error": "Task disappeared from background tasks"})
cleanup_background_task(task_id)
return f"Error: Task {task_id} disappeared from background tasks"
error = f"Task {task_id} disappeared from background tasks"
return _task_result_command(
tool_call_id=tool_call_id,
status="failed",
error=error,
)
# Log status changes for debugging
if result.status != last_status:
@ -398,28 +441,44 @@ async def task_tool(
writer({"type": "task_completed", "task_id": task_id, "result": result.result, "usage": usage})
logger.info(f"[trace={trace_id}] Task {task_id} completed after {poll_count} polls")
cleanup_background_task(task_id)
return f"Task Succeeded. Result: {result.result}"
return _task_result_command(
tool_call_id=tool_call_id,
status="completed",
result=result.result,
)
elif result.status == SubagentStatus.FAILED:
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
_report_subagent_usage(runtime, result)
writer({"type": "task_failed", "task_id": task_id, "error": result.error, "usage": usage})
logger.error(f"[trace={trace_id}] Task {task_id} failed: {result.error}")
cleanup_background_task(task_id)
return f"Task failed. Error: {result.error}"
return _task_result_command(
tool_call_id=tool_call_id,
status="failed",
error=result.error,
)
elif result.status == SubagentStatus.CANCELLED:
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
_report_subagent_usage(runtime, result)
writer({"type": "task_cancelled", "task_id": task_id, "error": result.error, "usage": usage})
logger.info(f"[trace={trace_id}] Task {task_id} cancelled: {result.error}")
cleanup_background_task(task_id)
return "Task cancelled by user."
return _task_result_command(
tool_call_id=tool_call_id,
status="cancelled",
error=result.error,
)
elif result.status == SubagentStatus.TIMED_OUT:
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
_report_subagent_usage(runtime, result)
writer({"type": "task_timed_out", "task_id": task_id, "error": result.error, "usage": usage})
logger.warning(f"[trace={trace_id}] Task {task_id} timed out: {result.error}")
cleanup_background_task(task_id)
return f"Task timed out. Error: {result.error}"
return _task_result_command(
tool_call_id=tool_call_id,
status="timed_out",
error=result.error,
)
# Still running, wait before next poll
await asyncio.sleep(5)
@ -440,7 +499,12 @@ async def task_tool(
# _background_tasks once the background thread reaches a terminal state.
request_cancel_background_task(task_id)
_schedule_deferred_subagent_cleanup(task_id, trace_id, max_poll_count)
return f"Task polling timed out after {timeout_minutes} minutes. This may indicate the background task is stuck. Status: {result.status.value}"
message = f"Task polling timed out after {timeout_minutes} minutes. This may indicate the background task is stuck. Status: {result.status.value}"
return _task_result_command(
tool_call_id=tool_call_id,
status="polling_timed_out",
error=message,
)
except asyncio.CancelledError:
# Signal the background subagent thread to stop cooperatively.
request_cancel_background_task(task_id)

5
backend/tests/test_create_deerflow_agent.py
View File

@ -308,9 +308,12 @@ def test_always_on_error_handling(mock_create_agent):
create_deerflow_agent(_make_mock_model(), features=feat)
call_kwargs = mock_create_agent.call_args[1]
mw_types = [type(m).__name__ for m in call_kwargs["middleware"]]
middleware = call_kwargs["middleware"]
mw_types = [type(m).__name__ for m in middleware]
assert "DanglingToolCallMiddleware" in mw_types
assert "ToolErrorHandlingMiddleware" in mw_types
tool_error_middleware = next(m for m in middleware if type(m).__name__ == "ToolErrorHandlingMiddleware")
assert tool_error_middleware._app_config is None
# ---------------------------------------------------------------------------

105
backend/tests/test_delegation_ledger.py
View File

@ -93,7 +93,16 @@ class TestExtractDelegations:
def test_completed_task_captured_with_result_metadata(self):
msgs = [
_ai_task_call("call_1", "research auth"),
ToolMessage(content="Task Succeeded. Result: auth uses JWT", tool_call_id="call_1", id="tm_1"),
ToolMessage(
content="Task Succeeded. Result: auth uses JWT",
tool_call_id="call_1",
id="tm_1",
additional_kwargs={
"subagent_status": "completed",
"subagent_result_brief": "auth uses JWT",
"subagent_result_sha256": "a" * 64,
},
),
]
out = extract_delegations(msgs)
@ -106,9 +115,9 @@ class TestExtractDelegations:
assert entry["status"] == "completed"
assert "auth uses JWT" in entry["result_brief"]
assert entry["result_ref"] == "tm_1"
assert len(entry["result_sha256"]) == 64
assert entry["result_sha256"] == "a" * 64
def test_status_kwarg_updates_dispatch(self):
def test_status_only_metadata_does_not_parse_result_from_content(self):
msgs = [
_ai_task_call("call_1", "research auth"),
ToolMessage(content="Task Succeeded. Result: ok", tool_call_id="call_1", additional_kwargs={"subagent_status": "completed"}),
@ -117,9 +126,62 @@ class TestExtractDelegations:
out = extract_delegations(msgs)
assert out[0]["status"] == "completed"
assert out[0]["result_brief"] == "ok"
assert "result_brief" not in out[0]
def test_falls_back_to_parsing_content_when_kwarg_absent(self):
def test_status_only_cancelled_metadata_keeps_terminal_detail_without_parsing_content(self):
msgs = [
_ai_task_call("call_cancelled", "stop task"),
ToolMessage(content="misleading content", tool_call_id="call_cancelled", id="tm_cancelled", additional_kwargs={"subagent_status": "cancelled"}),
]
out = extract_delegations(msgs)
assert out[0]["status"] == "cancelled"
assert out[0]["result_brief"] == "Task cancelled by user."
assert out[0]["result_ref"] == "tm_cancelled"
assert len(out[0]["result_sha256"]) == 64
def test_structured_result_metadata_wins_over_misleading_content(self):
msgs = [
_ai_task_call("call_1", "research auth"),
ToolMessage(
content="Task Succeeded. Result: misleading text",
tool_call_id="call_1",
id="tm_1",
additional_kwargs={
"subagent_status": "completed",
"subagent_result_brief": "structured text",
"subagent_result_sha256": "a" * 64,
},
),
]
out = extract_delegations(msgs)
assert out[0]["status"] == "completed"
assert out[0]["result_brief"] == "structured text"
assert out[0]["result_sha256"] == "a" * 64
def test_structured_error_metadata_wins_over_misleading_content(self):
msgs = [
_ai_task_call("call_2", "bad task"),
ToolMessage(
content="Task failed. Error: misleading boom",
tool_call_id="call_2",
id="tm_2",
additional_kwargs={
"subagent_status": "failed",
"subagent_error": "structured boom",
},
),
]
out = extract_delegations(msgs)
assert out[0]["status"] == "failed"
assert out[0]["result_brief"] == "structured boom"
def test_terminal_looking_content_without_structured_metadata_keeps_dispatch_in_progress(self):
msgs = [
_ai_task_call("call_2", "bad task"),
ToolMessage(content="Task failed. Error: boom", tool_call_id="call_2", id="tm_2"),
@ -127,13 +189,18 @@ class TestExtractDelegations:
out = extract_delegations(msgs)
assert out[0]["status"] == "failed"
assert "boom" in out[0]["result_brief"]
assert out[0]["status"] == "in_progress"
assert "result_brief" not in out[0]
def test_cancelled_task_status(self):
msgs = [
_ai_task_call("call_3", "cancelled task"),
ToolMessage(content="Task cancelled by user", tool_call_id="call_3", id="tm_3"),
ToolMessage(
content="Task cancelled by user",
tool_call_id="call_3",
id="tm_3",
additional_kwargs={"subagent_status": "cancelled", "subagent_error": "Task cancelled by user"},
),
]
out = extract_delegations(msgs)
@ -144,7 +211,12 @@ class TestExtractDelegations:
def test_timed_out_task_status(self):
msgs = [
_ai_task_call("call_timeout", "slow task"),
ToolMessage(content="Task timed out. Error: exceeded max runtime", tool_call_id="call_timeout", id="tm_timeout"),
ToolMessage(
content="Task timed out. Error: exceeded max runtime",
tool_call_id="call_timeout",
id="tm_timeout",
additional_kwargs={"subagent_status": "timed_out", "subagent_error": "exceeded max runtime"},
),
]
out = extract_delegations(msgs)
@ -159,6 +231,10 @@ class TestExtractDelegations:
content="Task polling timed out after 15 minutes. This may indicate the background task is stuck. Status: RUNNING",
tool_call_id="call_poll_timeout",
id="tm_poll_timeout",
additional_kwargs={
"subagent_status": "polling_timed_out",
"subagent_error": "Task polling timed out after 15 minutes. This may indicate the background task is stuck. Status: RUNNING",
},
),
]
@ -203,7 +279,16 @@ class TestExtractDelegations:
big = "x" * 10000
msgs = [
_ai_task_call("call_5", "big"),
ToolMessage(content=f"Task Succeeded. Result: {big}", tool_call_id="call_5", id="tm_5"),
ToolMessage(
content=f"Task Succeeded. Result: {big}",
tool_call_id="call_5",
id="tm_5",
additional_kwargs={
"subagent_status": "completed",
"subagent_result_brief": big[:2000],
"subagent_result_sha256": "b" * 64,
},
),
]
out = extract_delegations(msgs)

99
backend/tests/test_durable_context_middleware.py
View File

@ -1,17 +1,23 @@
from typing import Annotated
from _agent_e2e_helpers import FakeToolCallingModel
from langchain.agents import create_agent
from langchain.tools import InjectedToolCallId
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from deerflow.agents import thread_state as thread_state_module
from deerflow.agents.lead_agent import agent as lead_agent_module
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
from deerflow.agents.thread_state import ThreadState, merge_delegations
from deerflow.config.app_config import AppConfig
from deerflow.config.model_config import ModelConfig
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.subagents.status_contract import make_subagent_additional_kwargs
def _make_app_config() -> AppConfig:
@ -45,7 +51,12 @@ def _msgs_with_completed_task():
}
],
),
ToolMessage(content="Task Succeeded. Result: JWT", tool_call_id="call_1", id="tm_1"),
ToolMessage(
content="Task Succeeded. Result: JWT",
tool_call_id="call_1",
id="tm_1",
additional_kwargs=make_subagent_additional_kwargs("completed", result="JWT"),
),
]
@ -70,7 +81,12 @@ def _msgs_with_completed_tasks(count: int):
}
],
),
ToolMessage(content=f"Task Succeeded. Result: result {i}", tool_call_id=tool_call_id, id=f"tm_{i}"),
ToolMessage(
content=f"Task Succeeded. Result: result {i}",
tool_call_id=tool_call_id,
id=f"tm_{i}",
additional_kwargs=make_subagent_additional_kwargs("completed", result=f"result {i}"),
),
]
)
return messages
@ -155,6 +171,42 @@ class TestBeforeModelCapture:
assert out is None
def test_durable_context_uses_structured_task_metadata_when_content_disagrees(self):
middleware = DurableContextMiddleware()
state = {
"messages": [
AIMessage(
content="",
tool_calls=[
{
"name": "task",
"args": {
"description": "research",
"prompt": "do research",
"subagent_type": "general-purpose",
},
"id": "call-1",
"type": "tool_call",
}
],
),
ToolMessage(
content="Task Succeeded. Result: legacy",
tool_call_id="call-1",
additional_kwargs={
"subagent_status": "completed",
"subagent_result_brief": "structured",
"subagent_result_sha256": "a" * 64,
},
),
],
"delegations": [],
}
update = middleware.before_model(state, runtime=None)
assert update["delegations"][0]["result_brief"] == "structured"
class TestMiddlewareRegistration:
def test_registered_before_summarization(self, monkeypatch):
@ -189,7 +241,12 @@ class RecordingFakeModel(FakeToolCallingModel):
@tool("task", parse_docstring=True)
def fake_task(description: str, prompt: str, subagent_type: str) -> str:
def fake_task(
description: str,
prompt: str,
subagent_type: str,
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
"""Fake task tool.
Args:
@ -197,7 +254,18 @@ def fake_task(description: str, prompt: str, subagent_type: str) -> str:
prompt: full task instructions.
subagent_type: which subagent type to use.
"""
return "Task Succeeded. Result: AUTH_USES_JWT_SENTINEL"
return Command(
update={
"messages": [
ToolMessage(
content="Task Succeeded. Result: AUTH_USES_JWT_SENTINEL",
tool_call_id=tool_call_id,
name="task",
additional_kwargs=make_subagent_additional_kwargs("completed", result="AUTH_USES_JWT_SENTINEL"),
)
]
}
)
@tool("read_file", parse_docstring=True)
@ -314,6 +382,13 @@ class TestSkillContextCapture:
content="---\nname: data-analysis\ndescription: Analyze data.\n---\nBODY_SENTINEL",
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "data-analysis",
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": "Analyze data.",
}
},
),
]
@ -330,7 +405,18 @@ class TestSkillContextCapture:
middleware = DurableContextMiddleware(skills_container_path="/custom/skills", skill_file_read_tool_names=["open"])
msgs = [
AIMessage(content="", tool_calls=[{"name": "open", "args": {"path": "/custom/skills/public/x/SKILL.md"}, "id": "r1", "type": "tool_call"}]),
ToolMessage(content="---\nname: x\ndescription: d\n---\nbody", tool_call_id="r1", id="tm1"),
ToolMessage(
content="---\nname: x\ndescription: d\n---\nbody",
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "x",
"path": "/custom/skills/public/x/SKILL.md",
"description": "d",
}
},
),
]
out = middleware.before_model({"messages": msgs}, None)
@ -353,7 +439,7 @@ class TestSkillContextInjection:
agent = create_agent(
model=model,
tools=[fake_read_file],
middleware=[DurableContextMiddleware()],
middleware=[ToolErrorHandlingMiddleware(), DurableContextMiddleware()],
state_schema=ThreadState,
)
@ -381,6 +467,7 @@ class TestSkillContextInjection:
model=model,
tools=[fake_read_file],
middleware=[
ToolErrorHandlingMiddleware(),
DurableContextMiddleware(),
DeerFlowSummarizationMiddleware(model=summary_model, trigger=("messages", 4), keep=("messages", 2), token_counter=len),
],

39
backend/tests/test_gateway_imports.py Normal file
View File

@ -0,0 +1,39 @@
"""Gateway import regression tests."""
from __future__ import annotations
import os
import subprocess
import sys
from pathlib import Path
def test_gateway_app_imports_first_without_subagent_import_cycle() -> None:
"""The replay gateway imports app.gateway.app in a clean process."""
backend_root = Path(__file__).resolve().parents[1]
env = {**os.environ, "PYTHONPATH": str(backend_root)}
result = subprocess.run(
[sys.executable, "-c", "from app.gateway.app import app"],
capture_output=True,
text=True,
env=env,
)
assert result.returncode == 0, result.stderr
def test_subagent_package_public_executor_exports_are_lazy_importable() -> None:
"""The package-level executor exports must not re-enter their own import."""
backend_root = Path(__file__).resolve().parents[1]
env = {**os.environ, "PYTHONPATH": str(backend_root)}
result = subprocess.run(
[
sys.executable,
"-c",
"from deerflow.subagents import SubagentExecutor, SubagentResult; print(SubagentExecutor.__name__, SubagentResult.__name__)",
],
capture_output=True,
text=True,
env=env,
)
assert result.returncode == 0, result.stderr
assert "SubagentExecutor SubagentResult" in result.stdout

41
backend/tests/test_skill_container_path_defaults.py Normal file
View File

@ -0,0 +1,41 @@
"""Regression tests for the skills sandbox container root default."""
from __future__ import annotations
import ast
from pathlib import Path
def test_mnt_skills_literal_is_owned_by_skill_constants_module():
package_root = Path(__file__).parents[1] / "packages" / "harness" / "deerflow"
allowed = {package_root / "constants.py"}
offenders: list[str] = []
for path in package_root.rglob("*.py"):
if path in allowed:
continue
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and node.value == "/mnt/skills":
offenders.append(str(path.relative_to(package_root)))
assert offenders == []
def test_runtime_middlewares_use_top_level_skills_container_constant():
package_root = Path(__file__).parents[1] / "packages" / "harness" / "deerflow"
offenders: list[str] = []
for relative_path in (
Path("agents/middlewares/durable_context_middleware.py"),
Path("agents/middlewares/tool_error_handling_middleware.py"),
):
path = package_root / relative_path
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module == "deerflow.config.skills_config":
imported_names = {alias.name for alias in node.names}
if "DEFAULT_SKILLS_CONTAINER_PATH" in imported_names:
offenders.append(str(relative_path))
assert offenders == []

252
backend/tests/test_skill_context.py
View File

@ -1,6 +1,10 @@
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from deerflow.agents.middlewares.skill_context import extract_skills, render_skill_context
from deerflow.agents.middlewares.skill_context import (
build_skill_entry_metadata_from_read,
extract_skills,
render_skill_context,
)
_ROOT = "/mnt/skills"
_READ = frozenset({"read_file", "read", "view", "cat"})
@ -20,12 +24,44 @@ def _ai_read(tool_call_id: str, path: str, name: str = "read_file") -> AIMessage
)
def _skill_metadata(path: str = "/mnt/skills/public/data-analysis/SKILL.md", description: str = "Analyze data with pandas and charts.") -> dict:
return {
"skill_context_entry": {
"name": path.split("/")[-2],
"path": path,
"description": description,
}
}
class TestExtractSkills:
def test_build_skill_entry_metadata_from_read_rejects_non_skill_files(self):
assert (
build_skill_entry_metadata_from_read(
"/mnt/skills/public/data-analysis/README.md",
_SKILL_BODY,
skills_root=_ROOT,
)
is None
)
def test_build_skill_entry_metadata_from_read_returns_compact_reference(self):
entry = build_skill_entry_metadata_from_read(
"/mnt/skills/public/data-analysis/SKILL.md",
_SKILL_BODY,
skills_root=_ROOT,
)
assert entry == {
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": "Analyze data with pandas and charts.",
}
assert "ALWAYS_USE_PANDAS_SENTINEL" not in repr(entry)
def test_captures_skill_reference_with_description(self):
msgs = [
HumanMessage(content="use the analysis skill"),
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(content=_SKILL_BODY, tool_call_id="r1", id="tm1"),
ToolMessage(content=_SKILL_BODY, tool_call_id="r1", id="tm1", additional_kwargs=_skill_metadata()),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert len(out) == 1
@ -37,10 +73,15 @@ class TestExtractSkills:
assert isinstance(out[0]["loaded_at"], int)
def test_description_is_capped_at_capture_time(self):
body = "---\nname: huge\ndescription: " + ("x" * 2000) + "\n---\nBODY_SENTINEL"
description = "x" * 500
msgs = [
_ai_read("r1", "/mnt/skills/public/huge/SKILL.md"),
ToolMessage(content=body, tool_call_id="r1", id="tm1"),
ToolMessage(
content="BODY_SENTINEL",
tool_call_id="r1",
id="tm1",
additional_kwargs=_skill_metadata("/mnt/skills/public/huge/SKILL.md", description),
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
@ -49,26 +90,36 @@ class TestExtractSkills:
assert len(out[0]["description"]) <= 500
assert "BODY_SENTINEL" not in repr(out[0])
def test_missing_frontmatter_yields_empty_description(self):
def test_metadata_with_empty_description_yields_empty_description(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/x/SKILL.md"),
ToolMessage(content="# X\nno frontmatter here", tool_call_id="r1", id="tm1"),
ToolMessage(
content="# X\nno frontmatter here",
tool_call_id="r1",
id="tm1",
additional_kwargs=_skill_metadata("/mnt/skills/public/x/SKILL.md", ""),
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out and out[0]["description"] == ""
def test_malformed_frontmatter_yields_empty_description(self):
def test_missing_metadata_logs_warning_without_recovering_from_content(self, caplog):
msgs = [
_ai_read("r1", "/mnt/skills/public/x/SKILL.md"),
ToolMessage(content="---\n: : not valid yaml\n---\nbody", tool_call_id="r1", id="tm1"),
ToolMessage(content=_SKILL_BODY, tool_call_id="r1", id="tm1"),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out and out[0]["description"] == ""
with caplog.at_level("WARNING", logger="deerflow.agents.middlewares.skill_context"):
assert extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ) == []
assert "missing skill read metadata" in caplog.text
assert "tool_call_id=r1" in caplog.text
assert "/mnt/skills/public/x/SKILL.md" in caplog.text
def test_normalizes_dot_segments_under_skills_root(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/./data-analysis/SKILL.md"),
ToolMessage(content="body", tool_call_id="r1", id="tm1"),
ToolMessage(content="body", tool_call_id="r1", id="tm1", additional_kwargs=_skill_metadata()),
]
out = extract_skills(msgs, skills_root="/mnt/skills/", read_tool_names=_READ)
@ -129,7 +180,12 @@ class TestExtractSkills:
def test_trailing_slash_root_normalized(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/a/SKILL.md"),
ToolMessage(content="body", tool_call_id="r1", id="tm1"),
ToolMessage(
content="body",
tool_call_id="r1",
id="tm1",
additional_kwargs=_skill_metadata("/mnt/skills/public/a/SKILL.md", ""),
),
]
out = extract_skills(msgs, skills_root="/mnt/skills/", read_tool_names=_READ)
assert out and out[0]["name"] == "a"
@ -137,13 +193,179 @@ class TestExtractSkills:
def test_multiple_skills_each_captured(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/a/SKILL.md"),
ToolMessage(content="A", tool_call_id="r1", id="tm1"),
ToolMessage(
content="A",
tool_call_id="r1",
id="tm1",
additional_kwargs=_skill_metadata("/mnt/skills/public/a/SKILL.md", "A"),
),
_ai_read("r2", "/mnt/skills/custom/b/SKILL.md"),
ToolMessage(content="B", tool_call_id="r2", id="tm2"),
ToolMessage(
content="B",
tool_call_id="r2",
id="tm2",
additional_kwargs=_skill_metadata("/mnt/skills/custom/b/SKILL.md", "B"),
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert [e["name"] for e in out] == ["a", "b"]
def test_extract_skills_prefers_metadata_only_when_path_matches_read_call(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(
content="---\nname: wrong\ndescription: content body\n---\nbody",
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "data-analysis",
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": "Structured description.",
}
},
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out == [
{
"name": "data-analysis",
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": "Structured description.",
"loaded_at": 1,
}
]
def test_extract_skills_rejects_metadata_path_mismatch_without_reparsing_content(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(
content=_SKILL_BODY,
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "other",
"path": "/mnt/skills/public/other/SKILL.md",
"description": "Wrong metadata.",
}
},
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out == []
def test_extract_skills_warns_on_metadata_path_mismatch(self, caplog):
msgs = [
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(
content=_SKILL_BODY,
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "other",
"path": "/mnt/skills/public/other/SKILL.md",
"description": "Wrong metadata.",
}
},
),
]
with caplog.at_level("WARNING", logger="deerflow.agents.middlewares.skill_context"):
assert extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ) == []
assert "mismatched skill read metadata" in caplog.text
assert "expected_path=/mnt/skills/public/data-analysis/SKILL.md" in caplog.text
assert "metadata_path=/mnt/skills/public/other/SKILL.md" in caplog.text
def test_extract_skills_rebuilds_name_from_validated_read_path(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(
content="---\ndescription: content body\n---\nbody",
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "spoofed-name",
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": "Structured description.",
}
},
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out[0]["name"] == "data-analysis"
assert out[0]["path"] == "/mnt/skills/public/data-analysis/SKILL.md"
assert out[0]["description"] == "Structured description."
def test_extract_skills_accepts_same_path_metadata_with_missing_description(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(
content=_SKILL_BODY,
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "data-analysis",
"path": "/mnt/skills/public/data-analysis/SKILL.md",
}
},
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out[0]["path"] == "/mnt/skills/public/data-analysis/SKILL.md"
assert out[0]["description"] == ""
def test_extract_skills_accepts_same_path_metadata_with_non_string_description_as_empty(self):
msgs = [
_ai_read("r1", "/mnt/skills/public/data-analysis/SKILL.md"),
ToolMessage(
content=_SKILL_BODY,
tool_call_id="r1",
id="tm1",
additional_kwargs={
"skill_context_entry": {
"name": "data-analysis",
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": 123,
}
},
),
]
out = extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ)
assert out[0]["path"] == "/mnt/skills/public/data-analysis/SKILL.md"
assert out[0]["description"] == ""
def test_extract_skills_ignores_standalone_outside_root_metadata(self):
msgs = [
_ai_read("r1", "/workspace/notes.md"),
ToolMessage(
content="notes",
tool_call_id="r1",
additional_kwargs={
"skill_context_entry": {
"name": "secret",
"path": "/mnt/skills/public/secret/SKILL.md",
"description": "Do not trust this.",
}
},
),
]
assert extract_skills(msgs, skills_root=_ROOT, read_tool_names=_READ) == []
class TestRenderSkillContext:
def test_empty_returns_empty_string(self):
@ -172,7 +394,7 @@ class TestRenderSkillContext:
assert "- x" in out
assert "/mnt/skills/public/x/SKILL.md" in out
def test_render_caps_legacy_large_description(self):
def test_render_caps_large_description(self):
entries = [{"name": "x", "path": "/mnt/skills/public/x/SKILL.md", "description": "x" * 2000, "loaded_at": 0}]
out = render_skill_context(entries)

95
backend/tests/test_subagent_status_contract.py
View File

@ -1,29 +1,20 @@
"""Contract tests for ``deerflow.subagents.status_contract``.
Bytedance/deer-flow issue #3146: the backend stamps
``ToolMessage.additional_kwargs.subagent_status`` so the frontend can read
the subagent state from a structured field instead of parsing the result
text. The mapping from "task tool result text" to status is shared with the
frontend through the cross-language fixture file
``contracts/subagent_status_contract.json``.
These tests pin the backend implementation against that fixture so any
edit on either side surfaces immediately as a test failure.
"""
"""Contract tests for ``deerflow.subagents.status_contract``."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from deerflow.subagents.status_contract import (
SUBAGENT_ERROR_KEY,
SUBAGENT_METADATA_TEXT_MAX_CHARS,
SUBAGENT_RESULT_BRIEF_KEY,
SUBAGENT_RESULT_SHA256_KEY,
SUBAGENT_STATUS_KEY,
SUBAGENT_STATUS_VALUES,
extract_subagent_status,
_bound_metadata_text,
make_subagent_additional_kwargs,
read_subagent_result_metadata,
)
_REPO_ROOT = Path(__file__).resolve().parents[2]
@ -44,18 +35,6 @@ def test_status_values_match_contract():
assert set(SUBAGENT_STATUS_VALUES) == set(contract["valid_status_values"])
@pytest.mark.parametrize("case", _load_contract()["cases"], ids=lambda c: c["name"])
def test_extract_subagent_status_matches_contract(case):
"""Every fixture case maps through ``extract_subagent_status`` to the
expected status — covers task_tool's 5 normal returns, the 3
pre-execution ``Error:`` returns, the middleware-wrapped exception
case, whitespace handling, and the streaming chunk that must stay
unrecognised.
"""
status = extract_subagent_status(case["content"])
assert status == case["expected_status"], f"case {case['name']!r}: expected {case['expected_status']!r}, got {status!r}"
def test_make_subagent_additional_kwargs_includes_status():
kwargs = make_subagent_additional_kwargs("completed")
assert kwargs == {SUBAGENT_STATUS_KEY: "completed"}
@ -66,6 +45,30 @@ def test_make_subagent_additional_kwargs_includes_error_when_present():
assert kwargs == {SUBAGENT_STATUS_KEY: "failed", SUBAGENT_ERROR_KEY: "boom"}
def test_make_subagent_additional_kwargs_includes_bounded_result_metadata():
kwargs = make_subagent_additional_kwargs("completed", result="done")
assert kwargs[SUBAGENT_STATUS_KEY] == "completed"
assert kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "done"
assert len(kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64
assert SUBAGENT_ERROR_KEY not in kwargs
def test_make_subagent_additional_kwargs_bounds_large_result_metadata():
huge = "x" * (SUBAGENT_METADATA_TEXT_MAX_CHARS + 5000)
kwargs = make_subagent_additional_kwargs("completed", result=huge)
assert len(kwargs[SUBAGENT_RESULT_BRIEF_KEY]) <= SUBAGENT_METADATA_TEXT_MAX_CHARS
assert kwargs[SUBAGENT_RESULT_BRIEF_KEY] != huge
assert len(kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64
def test_bound_metadata_text_respects_small_caps():
text = "A" * 100
assert _bound_metadata_text(text, cap=0) == ""
assert _bound_metadata_text(text, cap=1) == "A"
assert len(_bound_metadata_text(text, cap=15)) <= 15
def test_make_subagent_additional_kwargs_omits_blank_error():
"""Empty / whitespace error must not leak as ``subagent_error: ""``."""
assert make_subagent_additional_kwargs("failed", error="") == {SUBAGENT_STATUS_KEY: "failed"}
@ -73,6 +76,44 @@ def test_make_subagent_additional_kwargs_omits_blank_error():
assert make_subagent_additional_kwargs("failed", error=None) == {SUBAGENT_STATUS_KEY: "failed"}
def test_make_subagent_additional_kwargs_bounds_large_error_metadata():
huge = "boom " * 2000
kwargs = make_subagent_additional_kwargs("failed", error=huge)
assert kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert len(kwargs[SUBAGENT_ERROR_KEY]) <= SUBAGENT_METADATA_TEXT_MAX_CHARS
assert SUBAGENT_RESULT_BRIEF_KEY not in kwargs
def test_read_subagent_result_metadata_returns_bounded_payload():
parsed = read_subagent_result_metadata(
{
SUBAGENT_STATUS_KEY: "completed",
SUBAGENT_RESULT_BRIEF_KEY: "structured",
SUBAGENT_RESULT_SHA256_KEY: "a" * 64,
SUBAGENT_ERROR_KEY: "ignored",
}
)
assert parsed == {
"status": "completed",
"result_brief": "structured",
"result_sha256": "a" * 64,
}
def test_read_subagent_result_metadata_rejects_unknown_status():
assert read_subagent_result_metadata({SUBAGENT_STATUS_KEY: "future"}) is None
def test_read_subagent_result_metadata_rejects_non_hex_sha256():
"""A 64-char value that is not a lowercase hex digest must be dropped."""
base = {SUBAGENT_STATUS_KEY: "completed", SUBAGENT_RESULT_BRIEF_KEY: "structured"}
for bad_hash in ("z" * 64, "A" * 64, "a" * 63, "a" * 65, ("a" * 63) + " "):
parsed = read_subagent_result_metadata({**base, SUBAGENT_RESULT_SHA256_KEY: bad_hash})
assert parsed == {"status": "completed", "result_brief": "structured"}, bad_hash
def test_make_subagent_additional_kwargs_rejects_unknown_status():
import pytest
with pytest.raises(ValueError, match="invalid subagent status"):
make_subagent_additional_kwargs("garbage") # type: ignore[arg-type]

240
backend/tests/test_task_tool_core_logic.py
View File

@ -2,13 +2,23 @@
import asyncio
import importlib
import inspect
from enum import Enum
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from deerflow.sandbox.security import LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE
from deerflow.subagents.config import SubagentConfig
from deerflow.subagents.status_contract import (
SUBAGENT_ERROR_KEY,
SUBAGENT_RESULT_BRIEF_KEY,
SUBAGENT_RESULT_SHA256_KEY,
SUBAGENT_STATUS_KEY,
)
# Use module import so tests can patch the exact symbols referenced inside task_tool().
task_tool_module = importlib.import_module("deerflow.tools.builtins.task_tool")
@ -71,7 +81,7 @@ def _make_result(
)
def _run_task_tool(**kwargs) -> str:
def _run_task_tool(**kwargs) -> str | Command:
"""Execute the task tool across LangChain sync/async wrapper variants."""
coroutine = getattr(task_tool_module.task_tool, "coroutine", None)
if coroutine is not None:
@ -79,6 +89,76 @@ def _run_task_tool(**kwargs) -> str:
return task_tool_module.task_tool.func(**kwargs)
def _task_tool_message(result: str | Command) -> ToolMessage:
assert isinstance(result, Command)
assert isinstance(result.update, dict)
messages = result.update["messages"]
assert len(messages) == 1
message = messages[0]
assert isinstance(message, ToolMessage)
return message
def test_task_result_command_derives_content_from_status_payload():
signature = inspect.signature(task_tool_module._task_result_command)
assert "content" not in signature.parameters
completed = _task_tool_message(
task_tool_module._task_result_command(
tool_call_id="tc-completed",
status="completed",
result="done",
)
)
assert completed.content == "Task Succeeded. Result: done"
assert completed.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed"
assert completed.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "done"
failed = _task_tool_message(
task_tool_module._task_result_command(
tool_call_id="tc-failed",
status="failed",
error="boom",
)
)
assert failed.content == "Task failed. Error: boom"
assert failed.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert failed.additional_kwargs[SUBAGENT_ERROR_KEY] == "boom"
failed_without_detail = _task_tool_message(
task_tool_module._task_result_command(
tool_call_id="tc-failed-empty",
status="failed",
error=None,
)
)
assert failed_without_detail.content == "Task failed."
assert failed_without_detail.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert failed_without_detail.additional_kwargs[SUBAGENT_ERROR_KEY] == "Task failed."
cancelled = _task_tool_message(
task_tool_module._task_result_command(
tool_call_id="tc-cancelled",
status="cancelled",
error=None,
)
)
assert cancelled.content == "Task cancelled by user."
assert cancelled.additional_kwargs[SUBAGENT_STATUS_KEY] == "cancelled"
assert cancelled.additional_kwargs[SUBAGENT_ERROR_KEY] == "Task cancelled by user."
timed_out_without_detail = _task_tool_message(
task_tool_module._task_result_command(
tool_call_id="tc-timeout-empty",
status="timed_out",
error="",
)
)
assert timed_out_without_detail.content == "Task timed out."
assert timed_out_without_detail.additional_kwargs[SUBAGENT_STATUS_KEY] == "timed_out"
assert timed_out_without_detail.additional_kwargs[SUBAGENT_ERROR_KEY] == "Task timed out."
async def _no_sleep(_: float) -> None:
return None
@ -100,7 +180,10 @@ def test_task_tool_returns_error_for_unknown_subagent(monkeypatch):
tool_call_id="tc-1",
)
assert result == "Error: Unknown subagent type 'general-purpose'. Available: general-purpose"
message = _task_tool_message(result)
assert message.content == "Task failed. Error: Unknown subagent type 'general-purpose'. Available: general-purpose"
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "Unknown subagent type 'general-purpose'. Available: general-purpose"
def test_task_tool_rejects_bash_subagent_when_host_bash_disabled(monkeypatch):
@ -115,7 +198,11 @@ def test_task_tool_rejects_bash_subagent_when_host_bash_disabled(monkeypatch):
tool_call_id="tc-bash",
)
assert result.startswith("Error: Bash subagent is disabled")
message = _task_tool_message(result)
assert isinstance(message.content, str)
assert message.content.startswith("Task failed. Error: Bash subagent is disabled")
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE
def test_task_tool_threads_runtime_app_config_to_subagent_dependencies(monkeypatch):
@ -171,7 +258,8 @@ def test_task_tool_threads_runtime_app_config_to_subagent_dependencies(monkeypat
tool_call_id="tc-explicit-config",
)
assert output == "Task Succeeded. Result: done"
message = _task_tool_message(output)
assert message.content == "Task Succeeded. Result: done"
assert captured["names_app_config"] is app_config
assert captured["config_lookup"] == ("bash", app_config)
assert captured["bash_gate_app_config"] is app_config
@ -227,7 +315,8 @@ def test_task_tool_emits_running_and_completed_events(monkeypatch):
tool_call_id="tc-123",
)
assert output == "Task Succeeded. Result: all done"
message = _task_tool_message(output)
assert message.content == "Task Succeeded. Result: all done"
assert captured["prompt"] == "collect diagnostics"
assert captured["task_id"] == "tc-123"
assert captured["executor_kwargs"]["thread_id"] == "thread-1"
@ -287,7 +376,7 @@ def test_task_tool_propagates_tool_groups_to_subagent(monkeypatch):
tool_call_id="tc-groups",
)
assert output == "Task Succeeded. Result: done"
assert _task_tool_message(output).content == "Task Succeeded. Result: done"
# The key assertion: groups should be propagated from parent metadata
get_available_tools.assert_called_once_with(model_name="ark-model", groups=parent_tool_groups, subagent_enabled=False)
@ -334,7 +423,7 @@ def test_task_tool_uses_subagent_model_override_for_tool_loading(monkeypatch):
tool_call_id="tc-issue-2543",
)
assert output == "Task Succeeded. Result: done"
assert _task_tool_message(output).content == "Task Succeeded. Result: done"
get_available_tools.assert_called_once_with(
model_name="vision-subagent-model",
groups=None,
@ -376,7 +465,7 @@ def test_task_tool_inherits_parent_skill_allowlist_for_default_subagent(monkeypa
tool_call_id="tc-skills",
)
assert output == "Task Succeeded. Result: done"
assert _task_tool_message(output).content == "Task Succeeded. Result: done"
assert captured["config"].skills == ["safe-skill"]
@ -422,7 +511,7 @@ def test_task_tool_intersects_parent_and_subagent_skill_allowlists(monkeypatch):
tool_call_id="tc-skills-intersection",
)
assert output == "Task Succeeded. Result: done"
assert _task_tool_message(output).content == "Task Succeeded. Result: done"
assert captured["config"].skills == ["safe-skill"]
@ -461,7 +550,7 @@ def test_task_tool_no_tool_groups_passes_none(monkeypatch):
tool_call_id="tc-no-groups",
)
assert output == "Task Succeeded. Result: ok"
assert _task_tool_message(output).content == "Task Succeeded. Result: ok"
# No tool_groups in metadata → groups=None (default behavior preserved)
get_available_tools.assert_called_once_with(model_name="ark-model", groups=None, subagent_enabled=False)
@ -501,7 +590,7 @@ def test_task_tool_runtime_none_passes_groups_none(monkeypatch):
tool_call_id="tc-no-runtime",
)
assert output == "Task Succeeded. Result: ok"
assert _task_tool_message(output).content == "Task Succeeded. Result: ok"
# runtime is None -> metadata is empty dict -> groups=None, model falls back to app default.
get_available_tools.assert_called_once_with(
model_name="default-model",
@ -538,7 +627,10 @@ def test_task_tool_runtime_none_passes_groups_none(monkeypatch):
tool_call_id="tc-fail",
)
assert output == "Task failed. Error: subagent crashed"
message = _task_tool_message(output)
assert message.content == "Task failed. Error: subagent crashed"
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "subagent crashed"
assert events[-1]["type"] == "task_failed"
assert events[-1]["error"] == "subagent crashed"
@ -572,7 +664,10 @@ def test_task_tool_returns_timed_out_message(monkeypatch):
tool_call_id="tc-timeout",
)
assert output == "Task timed out. Error: timeout"
message = _task_tool_message(output)
assert message.content == "Task timed out. Error: timeout"
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "timed_out"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "timeout"
assert events[-1]["type"] == "task_timed_out"
assert events[-1]["error"] == "timeout"
@ -608,7 +703,11 @@ def test_task_tool_polling_safety_timeout(monkeypatch):
tool_call_id="tc-safety-timeout",
)
assert output.startswith("Task polling timed out after 0 minutes")
message = _task_tool_message(output)
assert isinstance(message.content, str)
assert message.content.startswith("Task polling timed out after 0 minutes")
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "polling_timed_out"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == message.content
assert events[0]["type"] == "task_started"
assert events[-1]["type"] == "task_timed_out"
@ -649,7 +748,7 @@ def test_cleanup_called_on_completed(monkeypatch):
tool_call_id="tc-cleanup-completed",
)
assert output == "Task Succeeded. Result: done"
assert _task_tool_message(output).content == "Task Succeeded. Result: done"
assert cleanup_calls == ["tc-cleanup-completed"]
@ -689,7 +788,7 @@ def test_cleanup_called_on_failed(monkeypatch):
tool_call_id="tc-cleanup-failed",
)
assert output == "Task failed. Error: error"
assert _task_tool_message(output).content == "Task failed. Error: error"
assert cleanup_calls == ["tc-cleanup-failed"]
@ -729,7 +828,7 @@ def test_cleanup_called_on_timed_out(monkeypatch):
tool_call_id="tc-cleanup-timedout",
)
assert output == "Task timed out. Error: timeout"
assert _task_tool_message(output).content == "Task timed out. Error: timeout"
assert cleanup_calls == ["tc-cleanup-timedout"]
@ -792,7 +891,9 @@ def test_cleanup_not_called_on_polling_safety_timeout(monkeypatch):
tool_call_id="tc-no-cleanup-safety-timeout",
)
assert output.startswith("Task polling timed out after 0 minutes")
message = _task_tool_message(output)
assert isinstance(message.content, str)
assert message.content.startswith("Task polling timed out after 0 minutes")
# cleanup_background_task must NOT be called directly (task is still RUNNING)
assert cleanup_calls == []
# cooperative cancellation must be requested
@ -1077,11 +1178,112 @@ def test_task_tool_returns_cancelled_message(monkeypatch):
tool_call_id="tc-poll-cancelled",
)
assert output == "Task cancelled by user."
message = _task_tool_message(output)
assert message.content == "Task cancelled by user. Error: Cancelled by user"
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "cancelled"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "Cancelled by user"
assert any(e.get("type") == "task_cancelled" for e in events)
assert cleanup_calls == ["tc-poll-cancelled"]
def test_task_tool_emits_completed_metadata(monkeypatch):
config = _make_subagent_config()
monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
monkeypatch.setattr(
task_tool_module,
"SubagentExecutor",
type("DummyExecutor", (), {"__init__": lambda self, **kwargs: None, "execute_async": lambda self, prompt, task_id=None: task_id}),
)
monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: config)
monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"))
monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _: None)
monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
monkeypatch.setattr(task_tool_module, "_report_subagent_usage", lambda *_: None)
monkeypatch.setattr(task_tool_module, "cleanup_background_task", lambda _: None)
monkeypatch.setattr("deerflow.tools.get_available_tools", MagicMock(return_value=[]))
message = _task_tool_message(
_run_task_tool(
runtime=_make_runtime(),
description="执行任务",
prompt="do work",
subagent_type="general-purpose",
tool_call_id="tc-completed-metadata",
)
)
assert message.content == "Task Succeeded. Result: done"
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed"
assert message.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "done"
assert len(message.additional_kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64
def test_task_tool_emits_disappeared_task_metadata(monkeypatch):
config = _make_subagent_config()
events = []
monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
monkeypatch.setattr(
task_tool_module,
"SubagentExecutor",
type("DummyExecutor", (), {"__init__": lambda self, **kwargs: None, "execute_async": lambda self, prompt, task_id=None: task_id}),
)
monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: config)
monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: None)
monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: events.append)
monkeypatch.setattr(task_tool_module, "cleanup_background_task", lambda _: None)
monkeypatch.setattr("deerflow.tools.get_available_tools", MagicMock(return_value=[]))
message = _task_tool_message(
_run_task_tool(
runtime=_make_runtime(),
description="执行任务",
prompt="missing task",
subagent_type="general-purpose",
tool_call_id="tc-missing",
)
)
assert message.content == "Task failed. Error: Task tc-missing disappeared from background tasks"
assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "Task tc-missing disappeared from background tasks"
assert events[-1]["type"] == "task_failed"
def test_task_tool_bounds_large_result_metadata(monkeypatch):
config = _make_subagent_config()
huge = "x" * 10000
monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
monkeypatch.setattr(
task_tool_module,
"SubagentExecutor",
type("DummyExecutor", (), {"__init__": lambda self, **kwargs: None, "execute_async": lambda self, prompt, task_id=None: task_id}),
)
monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: config)
monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: _make_result(FakeSubagentStatus.COMPLETED, result=huge))
monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _: None)
monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
monkeypatch.setattr(task_tool_module, "_report_subagent_usage", lambda *_: None)
monkeypatch.setattr(task_tool_module, "cleanup_background_task", lambda _: None)
monkeypatch.setattr("deerflow.tools.get_available_tools", MagicMock(return_value=[]))
message = _task_tool_message(
_run_task_tool(
runtime=_make_runtime(),
description="执行任务",
prompt="large result",
subagent_type="general-purpose",
tool_call_id="tc-large-result",
)
)
assert message.content == f"Task Succeeded. Result: {huge}"
assert len(message.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY]) <= 2000
assert len(message.additional_kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64
def test_cancellation_reports_subagent_usage(monkeypatch):
"""Verify cancellation handler waits (shielded) for subagent terminal state,
then reports the final token usage before re-raising CancelledError.

98
backend/tests/test_tool_error_handling_middleware.py
View File

@ -11,10 +11,12 @@ from deerflow.agents.middlewares.tool_error_handling_middleware import (
build_subagent_runtime_middlewares,
)
from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
from deerflow.config import summarization_config
from deerflow.config.app_config import AppConfig, CircuitBreakerConfig
from deerflow.config.guardrails_config import GuardrailsConfig
from deerflow.config.model_config import ModelConfig
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.subagents.status_contract import SUBAGENT_ERROR_KEY, SUBAGENT_STATUS_KEY
def _request(name: str = "web_search", tool_call_id: str | None = "tc-1"):
@ -154,6 +156,21 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware
assert isinstance(middlewares[-1], SafetyFinishReasonMiddleware)
def test_lead_runtime_middlewares_thread_app_config_to_tool_error_handling(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setitem(
sys.modules,
"deerflow.agents.middlewares.input_sanitization_middleware",
_module("deerflow.agents.middlewares.input_sanitization_middleware", InputSanitizationMiddleware=object),
)
app_config = _make_app_config()
_stub_runtime_middleware_imports(monkeypatch)
middlewares = build_lead_runtime_middlewares(app_config=app_config)
tool_middleware = next(mw for mw in middlewares if isinstance(mw, ToolErrorHandlingMiddleware))
assert tool_middleware._app_config is app_config
def test_build_lead_runtime_middlewares_orders_thread_data_before_uploads():
"""ThreadDataMiddleware must run before UploadsMiddleware so the uploads
directory is guaranteed to exist when UploadsMiddleware scans it under
@ -233,6 +250,69 @@ def test_wrap_tool_call_passthrough_on_success():
assert result is expected
def test_read_file_skill_read_stamps_compact_skill_metadata():
app_config = _make_app_config()
app_config.skills.container_path = "/mnt/skills"
app_config.summarization.skill_file_read_tool_names = ["read_file"]
middleware = ToolErrorHandlingMiddleware(app_config=app_config)
req = _request(name="read_file", tool_call_id="read-1")
req.tool_call["args"] = {"path": "/mnt/skills/public/data-analysis/SKILL.md"}
result = middleware.wrap_tool_call(
req,
lambda _req: ToolMessage(
content="---\nname: data-analysis\ndescription: Analyze data.\n---\nBODY",
tool_call_id="read-1",
name="read_file",
),
)
assert result.additional_kwargs["skill_context_entry"] == {
"path": "/mnt/skills/public/data-analysis/SKILL.md",
"description": "Analyze data.",
}
def test_skill_read_config_is_cached_on_middleware_instance():
middleware = ToolErrorHandlingMiddleware()
default_names = getattr(summarization_config, "DEFAULT_SKILL_FILE_READ_TOOL_NAMES", None)
assert default_names is not None
assert middleware._skill_read_tool_names == frozenset(default_names)
assert middleware._skills_root == "/mnt/skills"
def test_skill_metadata_respects_custom_skills_root():
app_config = _make_app_config()
app_config.skills.container_path = "/custom/skills"
app_config.summarization.skill_file_read_tool_names = ["read_file"]
middleware = ToolErrorHandlingMiddleware(app_config=app_config)
req = _request(name="read_file", tool_call_id="read-1")
req.tool_call["args"] = {"path": "/custom/skills/public/x/SKILL.md"}
result = middleware.wrap_tool_call(
req,
lambda _req: ToolMessage("---\ndescription: X\n---\nBody", tool_call_id="read-1", name="read_file"),
)
assert result.additional_kwargs["skill_context_entry"]["path"] == "/custom/skills/public/x/SKILL.md"
def test_skill_metadata_disabled_when_read_tool_names_empty():
app_config = _make_app_config()
app_config.summarization.skill_file_read_tool_names = []
middleware = ToolErrorHandlingMiddleware(app_config=app_config)
req = _request(name="read_file", tool_call_id="read-1")
req.tool_call["args"] = {"path": "/mnt/skills/public/x/SKILL.md"}
result = middleware.wrap_tool_call(
req,
lambda _req: ToolMessage("---\ndescription: X\n---\nBody", tool_call_id="read-1", name="read_file"),
)
assert "skill_context_entry" not in result.additional_kwargs
def test_wrap_tool_call_returns_error_tool_message_on_exception():
middleware = ToolErrorHandlingMiddleware()
req = _request(name="web_search", tool_call_id="tc-42")
@ -250,6 +330,24 @@ def test_wrap_tool_call_returns_error_tool_message_on_exception():
assert "network down" in result.text
def test_task_exception_wrapper_uses_subagent_result_formatter():
middleware = ToolErrorHandlingMiddleware()
req = _request(name="task", tool_call_id="tc-task")
def _boom(_req):
raise RuntimeError("network down")
result = middleware.wrap_tool_call(req, _boom)
assert isinstance(result, ToolMessage)
assert result.tool_call_id == "tc-task"
assert result.name == "task"
assert result.status == "error"
assert result.content == "Task failed. Error: RuntimeError: network down. Continue with available context, or choose an alternative tool."
assert result.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed"
assert result.additional_kwargs[SUBAGENT_ERROR_KEY] == "RuntimeError: network down"
def test_wrap_tool_call_uses_fallback_tool_call_id_when_missing():
middleware = ToolErrorHandlingMiddleware()
req = _request(name="mcp_tool", tool_call_id=None)

137
backend/tests/test_tool_error_handling_subagent_stamp.py
View File

@ -1,25 +1,11 @@
"""Regression tests for ToolErrorHandlingMiddleware's subagent status stamp.
Bytedance/deer-flow issue #3146: rather than stamp
``ToolMessage.additional_kwargs.subagent_status`` from each of
task_tool.py's 5 normal returns + 3 pre-execution Error: returns (which
would be 8 separate places to drift over time), the middleware that
already wraps every tool call does the stamping in one place. These
tests pin that centralisation.
For non-``task`` tools the middleware must not touch additional_kwargs
— other tools have their own conventions and we do not want to leak a
``subagent_status`` field onto them.
"""
"""Tests for task exception metadata produced by ToolErrorHandlingMiddleware."""
from __future__ import annotations
import asyncio
import json
from pathlib import Path
import pytest
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from deerflow.agents.middlewares.tool_error_handling_middleware import (
ToolErrorHandlingMiddleware,
@ -29,14 +15,6 @@ from deerflow.subagents.status_contract import (
SUBAGENT_STATUS_KEY,
)
_CONTRACT_PATH = Path(__file__).resolve().parents[2] / "contracts" / "subagent_status_contract.json"
def _load_terminal_cases() -> list[dict]:
"""Load only the cases that should produce a terminal status stamp."""
data = json.loads(_CONTRACT_PATH.read_text(encoding="utf-8"))
return [c for c in data["cases"] if c["expected_status"] is not None]
class _FakeRequest:
"""Stand-in for ``ToolCallRequest`` used by the middleware."""
@ -45,51 +23,7 @@ class _FakeRequest:
self.tool_call = {"name": tool_name, "id": tool_call_id}
@pytest.mark.parametrize("case", _load_terminal_cases(), ids=lambda c: c["name"])
def test_stamps_subagent_status_on_successful_task_return(case):
"""Every terminal task tool result string stamps the matching status."""
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("task")
def handler(_req):
return ToolMessage(content=case["content"], tool_call_id="call-1", name="task")
result = middleware.wrap_tool_call(request, handler)
assert isinstance(result, ToolMessage)
assert result.additional_kwargs.get(SUBAGENT_STATUS_KEY) == case["expected_status"]
def test_does_not_stamp_unknown_streaming_chunk():
"""Non-terminal content leaves additional_kwargs alone."""
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("task")
def handler(_req):
return ToolMessage(content="Investigating ...", tool_call_id="call-1", name="task")
result = middleware.wrap_tool_call(request, handler)
assert SUBAGENT_STATUS_KEY not in (result.additional_kwargs or {})
def test_does_not_stamp_non_task_tool():
"""A non-task tool returning a string that happens to start with
``Error:`` must not pick up a subagent stamp."""
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("bash")
def handler(_req):
return ToolMessage(content="Error: command not found", tool_call_id="call-1", name="bash")
result = middleware.wrap_tool_call(request, handler)
assert SUBAGENT_STATUS_KEY not in (result.additional_kwargs or {})
def test_stamps_failed_when_task_tool_raises():
"""The exception path goes through ``_build_error_message`` which is
the only place ToolErrorHandlingMiddleware ever emits a brand-new
ToolMessage. It must stamp ``failed`` for task too, since the wrapper
text starts with ``Error:``.
"""
def test_task_tool_exception_returns_failed_metadata():
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("task")
@ -97,49 +31,72 @@ def test_stamps_failed_when_task_tool_raises():
raise RuntimeError("blew up during execution")
result = middleware.wrap_tool_call(request, handler)
assert isinstance(result, ToolMessage)
assert result.additional_kwargs.get(SUBAGENT_STATUS_KEY) == "failed"
assert "RuntimeError" in result.additional_kwargs.get(SUBAGENT_ERROR_KEY, "")
def test_async_wrap_also_stamps():
"""The async wrap path must behave identically."""
def test_async_task_tool_exception_returns_failed_metadata():
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("task")
async def handler(_req):
return ToolMessage(content="Task Succeeded. Result: ok", tool_call_id="call-1", name="task")
raise RuntimeError("async boom")
result = asyncio.run(middleware.awrap_tool_call(request, handler))
assert result.additional_kwargs.get(SUBAGENT_STATUS_KEY) == "completed"
assert isinstance(result, ToolMessage)
assert result.additional_kwargs.get(SUBAGENT_STATUS_KEY) == "failed"
assert "RuntimeError" in result.additional_kwargs.get(SUBAGENT_ERROR_KEY, "")
def test_preserves_existing_additional_kwargs():
"""The stamper must not clobber unrelated fields the tool already set."""
def test_successful_plain_task_tool_message_is_not_stamped_from_content():
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("task")
def handler(_req):
return ToolMessage(
content="Task Succeeded. Result: ok",
tool_call_id="call-1",
name="task",
additional_kwargs={"existing_field": "must_survive"},
)
return ToolMessage(content="Task Succeeded. Result: ok", tool_call_id="call-1", name="task")
result = middleware.wrap_tool_call(request, handler)
assert result.additional_kwargs.get("existing_field") == "must_survive"
assert result.additional_kwargs.get(SUBAGENT_STATUS_KEY) == "completed"
assert isinstance(result, ToolMessage)
assert SUBAGENT_STATUS_KEY not in (result.additional_kwargs or {})
def test_does_not_stamp_non_task_tool_exception():
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("bash")
def handler(_req):
raise RuntimeError("command failed")
result = middleware.wrap_tool_call(request, handler)
assert isinstance(result, ToolMessage)
assert SUBAGENT_STATUS_KEY not in (result.additional_kwargs or {})
def test_task_command_with_metadata_bypasses_middleware_stamp():
middleware = ToolErrorHandlingMiddleware()
request = _FakeRequest("task")
command = Command(
update={
"messages": [
ToolMessage(
"Task Succeeded. Result: ok",
tool_call_id="call-1",
name="task",
additional_kwargs={"subagent_status": "completed", "subagent_result_brief": "ok"},
)
]
}
)
assert middleware.wrap_tool_call(request, lambda _req: command) is command
def test_additional_kwargs_round_trip_via_json():
"""Pydantic dump → JSON → restore must keep the stamp intact.
``ToolMessage`` is what LangGraph serialises into the checkpoint and
what the frontend deserialises off the stream. If a future Pydantic /
LangChain upgrade silently strips unknown ``additional_kwargs`` we
want that to fail loudly here rather than in the wild.
"""
msg = ToolMessage(
content="Task Succeeded. Result: ok",
tool_call_id="call-1",

97
contracts/subagent_status_contract.json
View File

@ -1,98 +1,5 @@
{
"version": 1,
"description": "Cross-language contract test fixture for the subagent status field. The backend stamps ToolMessage.additional_kwargs.subagent_status using these prefixes; the frontend reads the structured field and falls back to the same prefixes. Both sides' tests load this file and must agree.",
"valid_status_values": ["completed", "failed", "cancelled", "timed_out", "polling_timed_out"],
"cases": [
{
"name": "succeeded",
"origin": "task_tool.py succeeded path",
"content": "Task Succeeded. Result: investigated and produced a 3-page report",
"expected_status": "completed",
"expected_error_contains": null
},
{
"name": "failed",
"origin": "task_tool.py failed path",
"content": "Task failed. Error: underlying tool raised RuntimeError",
"expected_status": "failed",
"expected_error_contains": "RuntimeError"
},
{
"name": "cancelled",
"origin": "task_tool.py cancelled path",
"content": "Task cancelled by user.",
"expected_status": "cancelled",
"expected_error_contains": null
},
{
"name": "timed_out",
"origin": "task_tool.py timed_out path",
"content": "Task timed out. Error: 900 seconds",
"expected_status": "timed_out",
"expected_error_contains": "900"
},
{
"name": "polling_timed_out",
"origin": "task_tool.py polling timeout safety-net path",
"content": "Task polling timed out after 15 minutes. This may indicate the background task is stuck. Status: RUNNING",
"expected_status": "polling_timed_out",
"expected_error_contains": "15"
},
{
"name": "polling_timed_out_other_n",
"origin": "varied N coverage",
"content": "Task polling timed out after 1 minutes. Status: RUNNING",
"expected_status": "polling_timed_out",
"expected_error_contains": null
},
{
"name": "pre_unknown_subagent",
"origin": "task_tool.py pre-execution Error path (unknown subagent type)",
"content": "Error: Unknown subagent type 'foo'. Available: bash, general-purpose",
"expected_status": "failed",
"expected_error_contains": "Unknown subagent"
},
{
"name": "pre_bash_disabled",
"origin": "task_tool.py pre-execution Error path (host bash disabled)",
"content": "Error: Host bash subagent is disabled by configuration",
"expected_status": "failed",
"expected_error_contains": "disabled"
},
{
"name": "pre_task_disappeared",
"origin": "task_tool.py pre-execution Error path (background task disappeared)",
"content": "Error: Task 1234 disappeared from background tasks",
"expected_status": "failed",
"expected_error_contains": "disappeared"
},
{
"name": "wrapper_error",
"origin": "ToolErrorHandlingMiddleware wrap on tool exception",
"content": "Error: Tool 'task' failed with TypeError: 'AsyncCallbackManager' object is not iterable. Continue with available context, or choose an alternative tool.",
"expected_status": "failed",
"expected_error_contains": "TypeError"
},
{
"name": "streaming_chunk_unknown",
"origin": "non-terminal chunk reaching parser",
"content": "Investigating ...",
"expected_status": null,
"expected_error_contains": null
},
{
"name": "succeeded_with_surrounding_whitespace",
"origin": "streaming sometimes prepends/appends newlines",
"content": " Task Succeeded. Result: ok ",
"expected_status": "completed",
"expected_error_contains": "ok"
},
{
"name": "cancelled_with_surrounding_whitespace",
"origin": "streaming whitespace coverage",
"content": " Task cancelled by user.\n",
"expected_status": "cancelled",
"expected_error_contains": null
}
]
"description": "Cross-language contract fixture for the structured subagent status field. Task result text is display content only and is not part of this wire contract.",
"valid_status_values": ["completed", "failed", "cancelled", "timed_out", "polling_timed_out"]
}

2
frontend/src/content/en/harness/middlewares.mdx
View File

@ -135,6 +135,8 @@ Audits sandbox operations performed during the agent's execution. Provides a rec
Captures long-lived runtime facts into explicit thread-state channels before summarization compacts the raw transcript. It records compressed summaries, delegated task state/results, and loaded `SKILL.md` references, then projects them into later model calls as hidden durable context data.
The backend stamps bounded structured task-result and skill-read metadata before durable context capture. The frontend task card reads the structured task status/result fields; task result text remains display content and is not parsed as the wire protocol.
**Configuration**: built in. `summarization.skill_file_read_tool_names` controls which read tools count as skill-reference reads; set it to `[]` to disable durable skill-reference capture.
---

2
frontend/src/content/zh/harness/middlewares.mdx
View File

@ -127,6 +127,8 @@ token_usage:
在摘要压缩原始 transcript 之前，将长期运行事实捕获到明确的线程状态 channel 中。它记录压缩摘要、已委托任务的状态/结果，以及已加载的 `SKILL.md` 引用，并在后续模型调用中作为隐藏 durable context 数据投影进去。
后端会在 durable context 捕获之前写入有界的结构化 task 结果和技能读取元数据。前端 task card 读取结构化 task status/result 字段；task 结果文本只作为展示内容，不再作为 wire protocol 解析。
**配置**：内置。`summarization.skill_file_read_tool_names` 控制哪些读取工具会被视为技能引用读取；设为 `[]` 可关闭 durable 技能引用捕获。
---

168
frontend/src/core/tasks/subtask-result.ts
View File

@ -16,12 +16,31 @@ export interface SubtaskResultUpdate {
*
* The values mirror the Python contract in
* ``backend/packages/harness/deerflow/subagents/status_contract.py``
* (``SUBAGENT_STATUS_KEY`` / ``SUBAGENT_ERROR_KEY``). The cross-language
* fixture at ``contracts/subagent_status_contract.json`` pins both sides
* to the same values.
* (``SUBAGENT_STATUS_KEY`` / ``SUBAGENT_ERROR_KEY`` /
* ``SUBAGENT_RESULT_BRIEF_KEY`` / ``SUBAGENT_RESULT_SHA256_KEY``). The
* result metadata fields are optional and bounded: ``subagent_result_brief``
* carries a trimmed summary for completed tasks and
* ``subagent_result_sha256`` carries the full-result digest. The
* cross-language fixture at ``contracts/subagent_status_contract.json``
* pins both sides to the same values.
*/
export const SUBAGENT_STATUS_KEY = "subagent_status";
export const SUBAGENT_ERROR_KEY = "subagent_error";
export const SUBAGENT_RESULT_BRIEF_KEY = "subagent_result_brief";
export const SUBAGENT_RESULT_SHA256_KEY = "subagent_result_sha256";
const STRUCTURED_SUBAGENT_KEYS = [
SUBAGENT_STATUS_KEY,
SUBAGENT_ERROR_KEY,
SUBAGENT_RESULT_BRIEF_KEY,
SUBAGENT_RESULT_SHA256_KEY,
];
const SUCCESS_PREFIX = "Task Succeeded. Result:";
const FAILURE_PREFIX = "Task failed.";
const TIMEOUT_PREFIX = "Task timed out";
const CANCELLED_PREFIX = "Task cancelled by user.";
const POLLING_TIMEOUT_PREFIX = "Task polling timed out";
const ERROR_WRAPPER_PATTERN = /^Error\b/i;
/**
* Map from the backend ``subagent_status`` value to the frontend
@ -39,53 +58,15 @@ const STRUCTURED_STATUS_TO_SUBTASK: Record<string, SubtaskStatus> = {
polling_timed_out: "failed",
};
/**
* Prefix strings the backend `task` tool writes into its result `content`.
*
* These values are not user-facing copy — they are part of the
* backend↔frontend contract defined in
* `backend/packages/harness/deerflow/tools/builtins/task_tool.py` (returned
* from the tool body) and in
* `backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py`
* (wrapper for tool exceptions). Any change here must be paired with the
* matching backend change. Exported so a future structured-status migration
* can reference the same values from one place.
*
* `task_tool.py` also emits three `Error:` strings for pre-execution failures
* — unknown subagent type, host-bash disabled, and "task disappeared from
* background tasks". They are handled by {@link ERROR_WRAPPER_PATTERN}
* rather than dedicated prefixes because the wrapper already produces
* exactly the right `terminal failed` shape.
*/
export const SUCCESS_PREFIX = "Task Succeeded. Result:";
export const FAILURE_PREFIX = "Task failed.";
export const TIMEOUT_PREFIX = "Task timed out";
export const CANCELLED_PREFIX = "Task cancelled by user.";
export const POLLING_TIMEOUT_PREFIX = "Task polling timed out";
export const ERROR_WRAPPER_PATTERN = /^Error\b/i;
/**
* Map a `task` tool result to a {@link SubtaskStatus}.
*
* Bytedance/deer-flow issue #3146: prefers the structured
* ``additional_kwargs.subagent_status`` field the backend now stamps via
* ``ToolErrorHandlingMiddleware``. Falls back to the legacy prefix
* matching for messages that pre-date the stamping commit (historical
* threads, third-party clients, or any tool path that bypasses the
* middleware). Both shapes converge on the same {@link SubtaskStatus}
* vocabulary the card UI renders.
* The backend writes task lifecycle facts into
* ``ToolMessage.additional_kwargs``. The textual ``content`` remains
* model-visible display content only; it is not parsed as a protocol.
*
* When the structured field is present, the prefix parser is still run
* so the success `result` body and the wrapped-error message can be
* back-filled from `content`. Today the backend only stamps the
* `subagent_status` enum value — the human-facing payload still lives
* in `content`, so dropping the prefix parse would regress the subtask
* card display. Structured fields win on conflict: if `subagent_status`
* and the text disagree, the text-derived `result`/`error` are
* discarded so a malformed wrapper can't sneak through.
*
* Returning `in_progress` is the **deliberate** fallback for content that
* matches none of the known prefixes and carries no structured stamp.
* Returning `in_progress` is the **deliberate** default for content that
* carries no structured stamp.
* LangChain only ever emits a `ToolMessage` once the tool itself has
* returned (success or wrapped exception), so an unknown shape means
* "the contract changed underneath us" — surfacing it as still-running
@ -96,60 +77,26 @@ export function parseSubtaskResult(
text: string,
additionalKwargs?: Record<string, unknown> | null,
): SubtaskResultUpdate {
const fromText = parseFromText(text.trim());
const structured = readStructuredStatus(additionalKwargs);
if (!structured) {
return fromText;
if (!hasStructuredSubagentMetadata(additionalKwargs)) {
return parseLegacyTaskResult(text.trim());
}
return { status: "in_progress" };
}
const update: SubtaskResultUpdate = { status: structured.status };
// Structured `subagent_error` wins; otherwise inherit the text-derived
// error only when both sides agree on the status (so a "Task Succeeded"
// body can't bleed into a `failed` structured stamp and vice versa).
if (structured.error) {
update.error = structured.error;
} else if (
fromText.status === structured.status &&
fromText.error !== undefined
) {
update.error = fromText.error;
}
// Result body only matters for `completed`; require text agreement so
// a lying success prefix under a `failed` stamp is dropped.
if (
structured.status === "completed" &&
fromText.status === "completed" &&
fromText.result !== undefined
) {
update.result = fromText.result;
const structuredResult = readStructuredResultBrief(additionalKwargs);
if (structured.status === "completed" && structuredResult) {
update.result = structuredResult;
}
return update;
}
export function hasSubtaskToolResult(
toolCallId: string | undefined,
messages: Message[],
) {
if (!toolCallId) {
return false;
}
return messages.some(
(message) => message.type === "tool" && message.tool_call_id === toolCallId,
);
}
export function derivePendingSubtaskStatus(
toolCallId: string | undefined,
messages: Message[],
isCurrentTurnLoading: boolean,
): SubtaskStatus {
if (isCurrentTurnLoading || hasSubtaskToolResult(toolCallId, messages)) {
return "in_progress";
}
return "failed";
}
function parseFromText(trimmed: string): SubtaskResultUpdate {
function parseLegacyTaskResult(trimmed: string): SubtaskResultUpdate {
if (trimmed.startsWith(SUCCESS_PREFIX)) {
return {
status: "completed",
@ -176,8 +123,6 @@ function parseFromText(trimmed: string): SubtaskResultUpdate {
return { status: "failed", error: trimmed };
}
// ToolErrorHandlingMiddleware-style wrapper, or any other terminal error
// signal the backend forwards to the lead agent.
if (ERROR_WRAPPER_PATTERN.test(trimmed)) {
return { status: "failed", error: trimmed };
}
@ -185,6 +130,29 @@ function parseFromText(trimmed: string): SubtaskResultUpdate {
return { status: "in_progress" };
}
export function hasSubtaskToolResult(
toolCallId: string | undefined,
messages: Message[],
) {
if (!toolCallId) {
return false;
}
return messages.some(
(message) => message.type === "tool" && message.tool_call_id === toolCallId,
);
}
export function derivePendingSubtaskStatus(
toolCallId: string | undefined,
messages: Message[],
isCurrentTurnLoading: boolean,
): SubtaskStatus {
if (isCurrentTurnLoading || hasSubtaskToolResult(toolCallId, messages)) {
return "in_progress";
}
return "failed";
}
interface StructuredStatus {
status: SubtaskStatus;
error?: string;
@ -198,10 +166,6 @@ function readStructuredStatus(
if (typeof rawStatus !== "string") return null;
const mapped = STRUCTURED_STATUS_TO_SUBTASK[rawStatus];
if (mapped === undefined) {
// Unknown future status value — stay on the legacy prefix fallback
// so a backend that ships a new enum variant before the frontend
// upgrades still renders something predictable instead of getting
// pinned to "in_progress" by an empty branch.
return null;
}
const rawError = additionalKwargs[SUBAGENT_ERROR_KEY];
@ -211,3 +175,19 @@ function readStructuredStatus(
}
return result;
}
function hasStructuredSubagentMetadata(
additionalKwargs: Record<string, unknown> | null | undefined,
): boolean {
if (!additionalKwargs) return false;
return STRUCTURED_SUBAGENT_KEYS.some((key) =>
Object.prototype.hasOwnProperty.call(additionalKwargs, key),
);
}
function readStructuredResultBrief(
additionalKwargs: Record<string, unknown> | null | undefined,
): string | undefined {
const value = additionalKwargs?.[SUBAGENT_RESULT_BRIEF_KEY];
return typeof value === "string" && value.trim() ? value.trim() : undefined;
}

241
frontend/tests/unit/core/tasks/subtask-result.test.ts
View File

@ -6,22 +6,15 @@ import { describe, expect, it } from "@rstest/core";
import {
SUBAGENT_ERROR_KEY,
SUBAGENT_RESULT_BRIEF_KEY,
SUBAGENT_STATUS_KEY,
derivePendingSubtaskStatus,
hasSubtaskToolResult,
parseSubtaskResult,
} from "@/core/tasks/subtask-result";
interface ContractCase {
name: string;
content: string;
expected_status: string | null;
expected_error_contains: string | null;
}
interface ContractFile {
valid_status_values: string[];
cases: ContractCase[];
}
const CONTRACT_PATH = resolve(
@ -33,112 +26,59 @@ const CONTRACT: ContractFile = JSON.parse(
) as ContractFile;
describe("parseSubtaskResult", () => {
it("recognises the standard success prefix", () => {
const parsed = parseSubtaskResult(
"Task Succeeded. Result: investigated and produced a 3-page report",
);
expect(parsed.status).toBe("completed");
expect(parsed.result).toBe("investigated and produced a 3-page report");
});
it("recognises the standard failure prefix", () => {
const parsed = parseSubtaskResult(
"Task failed. underlying tool raised RuntimeError",
);
expect(parsed.status).toBe("failed");
expect(parsed.error).toBe("underlying tool raised RuntimeError");
});
it("recognises the standard timeout prefix", () => {
const parsed = parseSubtaskResult("Task timed out after 900s");
expect(parsed.status).toBe("failed");
expect(parsed.error).toBe("Task timed out after 900s");
});
it("recognises the cancelled-by-user prefix", () => {
// bytedance/deer-flow#3131 review: this is one of the five terminal
// strings task_tool.py actually emits — the previous cut treated it as
// unrecognised content and pushed the card back to in_progress.
const parsed = parseSubtaskResult("Task cancelled by user.");
expect(parsed.status).toBe("failed");
expect(parsed.error).toBe("Task cancelled by user.");
});
it("recognises the polling-timed-out prefix", () => {
// Emitted by task_tool when the background polling loop runs out of
// budget waiting for the subagent to reach a terminal state.
const parsed = parseSubtaskResult(
"Task polling timed out after 15 minutes. This may indicate the background task is stuck. Status: RUNNING",
);
expect(parsed.status).toBe("failed");
expect(parsed.error).toContain("polling timed out");
});
it("recognises polling-timed-out with different durations", () => {
// `task_tool` emits `Task polling timed out after {N} minutes` where N
// varies with the configured subagent timeout. Guard against the regex
// accidentally being pinned to a specific number.
for (const n of [1, 5, 60]) {
const parsed = parseSubtaskResult(
`Task polling timed out after ${n} minutes. Status: RUNNING`,
);
expect(parsed.status).toBe("failed");
}
});
it("trims whitespace around cancelled and polling-timed-out prefixes", () => {
// Streaming chunks sometimes arrive with leading/trailing newlines.
expect(parseSubtaskResult(" Task cancelled by user. \n").status).toBe(
"failed",
);
it("uses legacy task result text when structured metadata is absent", () => {
expect(
parseSubtaskResult("\n\nTask polling timed out after 3 minutes").status,
).toBe("failed");
parseSubtaskResult(
"Task Succeeded. Result: investigated and produced a 3-page report",
),
).toEqual({
status: "completed",
result: "investigated and produced a 3-page report",
});
expect(
parseSubtaskResult(
"Task failed. Error: underlying tool raised RuntimeError",
),
).toEqual({
status: "failed",
error: "Error: underlying tool raised RuntimeError",
});
expect(parseSubtaskResult("Task cancelled by user.")).toEqual({
status: "failed",
error: "Task cancelled by user.",
});
expect(parseSubtaskResult("Task timed out. Error: 900 seconds")).toEqual({
status: "failed",
error: "Task timed out. Error: 900 seconds",
});
expect(
parseSubtaskResult(
"Task polling timed out after 15 minutes. Status: RUNNING",
),
).toEqual({
status: "failed",
error: "Task polling timed out after 15 minutes. Status: RUNNING",
});
expect(
parseSubtaskResult("Error: Tool 'task' failed with TypeError: boom"),
).toEqual({
status: "failed",
error: "Error: Tool 'task' failed with TypeError: boom",
});
});
it("recognises task_tool pre-execution Error: returns via the wrapper", () => {
// `task_tool.py` returns three `Error:` strings for unknown subagent
// type, host-bash disabled, and "task disappeared". They share the
// ERROR_WRAPPER_PATTERN, not a dedicated prefix, so this guards
// against a refactor splitting them off.
for (const text of [
"Error: Unknown subagent type 'foo'. Available: bash, general-purpose",
"Error: Host bash subagent is disabled by configuration",
"Error: Task 1234 disappeared from background tasks",
]) {
expect(parseSubtaskResult(text).status).toBe("failed");
}
});
it("keeps unknown content-only task results in progress", () => {
const parsed = parseSubtaskResult("partial streaming chunk");
it("treats middleware-wrapped tool errors as terminal failures", () => {
// bytedance/deer-flow issue #3107 BUG-007: the parent-visible ToolMessage
// produced by ToolErrorHandlingMiddleware never matches the three legacy
// prefixes, so subtask cards stay stuck on "in_progress".
const parsed = parseSubtaskResult(
"Error: Tool 'task' failed with TypeError: 'AsyncCallbackManager' object is not iterable. Continue with available context, or choose an alternative tool.",
);
expect(parsed.status).toBe("failed");
expect(parsed.error).toContain("AsyncCallbackManager");
});
it("treats any other Error: prefix as a terminal failure", () => {
const parsed = parseSubtaskResult("Error: subagent worker pool exhausted");
expect(parsed.status).toBe("failed");
});
it("keeps unrecognised non-error output as in_progress", () => {
// Streaming partial chunks should not flip the card to terminal early.
const parsed = parseSubtaskResult("Investigating ...");
expect(parsed.status).toBe("in_progress");
expect(parsed.error).toBeUndefined();
expect(parsed.result).toBeUndefined();
});
it("trims surrounding whitespace before matching prefixes", () => {
const parsed = parseSubtaskResult(" Task Succeeded. Result: ok ");
expect(parsed.status).toBe("completed");
expect(parsed.result).toBe("ok");
});
});
describe("hasSubtaskToolResult", () => {
@ -236,62 +176,57 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)",
expect(parsed.error).toBeUndefined();
});
it("falls back to prefix parsing when the structured status is missing", () => {
it("ignores terminal-looking content when partial structured metadata is present", () => {
const parsed = parseSubtaskResult("Task Succeeded. Result: foo", {
// No subagent_status here — backend versions that pre-date the
// middleware stamping commit still need to render.
other_field: "irrelevant",
[SUBAGENT_RESULT_BRIEF_KEY]: "structured result without status",
});
expect(parsed.status).toBe("completed");
expect(parsed.result).toBe("foo");
expect(parsed.status).toBe("in_progress");
expect(parsed.result).toBeUndefined();
});
it("falls back to prefix parsing when the structured status is an unknown future value", () => {
it("ignores terminal-looking content when the structured status is unknown", () => {
const parsed = parseSubtaskResult("Task Succeeded. Result: foo", {
[SUBAGENT_STATUS_KEY]: "renamed_in_v3",
});
// Falls back to prefix and still finds the success path.
expect(parsed.status).toBe("completed");
expect(parsed.status).toBe("in_progress");
});
it("structured status overrides legacy text — opposite content", () => {
// Defence: if backend sends `failed` structured but the content
// accidentally starts with "Task Succeeded.", we must trust the
// structured field. The structured field is the source of truth.
it("structured status overrides misleading content", () => {
const parsed = parseSubtaskResult("Task Succeeded. Result: this is a lie", {
[SUBAGENT_STATUS_KEY]: "failed",
});
expect(parsed.status).toBe("failed");
// The misleading success body must be dropped — `result` is reserved
// for the completed pill, and the suspicious text isn't replayed as
// an error either.
expect(parsed.result).toBeUndefined();
expect(parsed.error).toBeUndefined();
});
it("back-fills `result` from the success-prefixed content when structured says completed", () => {
// The backend currently stamps `subagent_status: completed` but the
// success body still lives in `content`. Without back-fill the card
// would render an empty completed pill (regression flagged in PR #3154
// Copilot review).
const parsed = parseSubtaskResult(
"Task Succeeded. Result: investigated and produced a 3-page report",
{ [SUBAGENT_STATUS_KEY]: "completed" },
);
it("does not back-fill result from content when structured result metadata is missing", () => {
const parsed = parseSubtaskResult("Task Succeeded. Result: text-only", {
[SUBAGENT_STATUS_KEY]: "completed",
});
expect(parsed.status).toBe("completed");
expect(parsed.result).toBe("investigated and produced a 3-page report");
expect(parsed.result).toBeUndefined();
});
it("back-fills `error` from a wrapped-error body when structured says failed and no subagent_error", () => {
// Same regression on the failure side: the wrapper text is the only
// place the diagnostic message exists when the backend stamps the
// enum but not `subagent_error`.
it("uses bounded structured result metadata when present for completed task", () => {
const parsed = parseSubtaskResult("Task Succeeded. Result: text body", {
[SUBAGENT_STATUS_KEY]: "completed",
subagent_result_brief: "structured",
subagent_result_sha256: "a".repeat(64),
});
expect(parsed.status).toBe("completed");
expect(parsed.result).toBe("structured");
});
it("does not back-fill error from content when structured error metadata is missing", () => {
const parsed = parseSubtaskResult(
"Error: Tool 'task' failed with TypeError: boom",
{ [SUBAGENT_STATUS_KEY]: "failed" },
{
[SUBAGENT_STATUS_KEY]: "failed",
},
);
expect(parsed.status).toBe("failed");
expect(parsed.error).toContain("TypeError: boom");
expect(parsed.error).toBeUndefined();
});
it("leaves `error` undefined when structured says failed with no error and unrecognised text", () => {
@ -306,32 +241,22 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)",
});
/**
* Cross-language contract test (bytedance/deer-flow#3146).
*
* Loads the shared fixture at ``contracts/subagent_status_contract.json``
* and runs every case through the legacy prefix parser. The matching
* backend test (`backend/tests/test_subagent_status_contract.py`) runs
* the same cases through ``extract_subagent_status``. Any drift between
* the two implementations surfaces here.
*
* Status-collapse expectations:
* - `completed` → `completed`
* - `failed` → `failed`
* - `cancelled` / `timed_out` / `polling_timed_out` → `failed`
* (the frontend card has three pill states, not five)
* - `null` → `in_progress`
* Cross-language contract test for the structured subagent status field.
* The backend and frontend share the enum values, but task result text is
* no longer part of the wire contract.
*/
describe("parseSubtaskResult — shared contract fixture", () => {
const expectedCardStatus = (backendStatus: string | null): string => {
if (backendStatus === null) return "in_progress";
const expectedCardStatus = (backendStatus: string): string => {
if (backendStatus === "completed") return "completed";
return "failed";
};
for (const c of CONTRACT.cases) {
it(`legacy prefix parser matches contract: ${c.name}`, () => {
const parsed = parseSubtaskResult(c.content);
expect(parsed.status).toBe(expectedCardStatus(c.expected_status));
for (const status of CONTRACT.valid_status_values) {
it(`maps structured status: ${status}`, () => {
const parsed = parseSubtaskResult("ignored content", {
[SUBAGENT_STATUS_KEY]: status,
});
expect(parsed.status).toBe(expectedCardStatus(status));
});
}
});