diff --git a/backend/AGENTS.md b/backend/AGENTS.md index fa5b5c040..1dd796801 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -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): diff --git a/backend/docs/summarization.md b/backend/docs/summarization.md index 71c51cb7a..582c7ef65 100644 --- a/backend/docs/summarization.md +++ b/backend/docs/summarization.md @@ -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 diff --git a/backend/packages/harness/deerflow/agents/__init__.py b/backend/packages/harness/deerflow/agents/__init__.py index 397f67f8e..4b56cbc06 100644 --- a/backend/packages/harness/deerflow/agents/__init__.py +++ b/backend/packages/harness/deerflow/agents/__init__.py @@ -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}") diff --git a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py index f94a07014..c711875c6 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py +++ b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py @@ -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 diff --git a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py index a99b5a30d..8122f66bd 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py +++ b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py @@ -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] diff --git a/backend/packages/harness/deerflow/agents/middlewares/durable_context_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/durable_context_middleware.py index f107bc9e4..fca49ccfe 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/durable_context_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/durable_context_middleware.py @@ -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: diff --git a/backend/packages/harness/deerflow/agents/middlewares/skill_context.py b/backend/packages/harness/deerflow/agents/middlewares/skill_context.py index c98d1c06f..75163148e 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/skill_context.py +++ b/backend/packages/harness/deerflow/agents/middlewares/skill_context.py @@ -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, } ) diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index 6cc37acf1..8e2bae352 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -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] diff --git a/backend/packages/harness/deerflow/config/skills_config.py b/backend/packages/harness/deerflow/config/skills_config.py index c306ce747..158ec8691 100644 --- a/backend/packages/harness/deerflow/config/skills_config.py +++ b/backend/packages/harness/deerflow/config/skills_config.py @@ -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", ) diff --git a/backend/packages/harness/deerflow/config/summarization_config.py b/backend/packages/harness/deerflow/config/summarization_config.py index 571fb4296..d59e841a1 100644 --- a/backend/packages/harness/deerflow/config/summarization_config.py +++ b/backend/packages/harness/deerflow/config/summarization_config.py @@ -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.", ) diff --git a/backend/packages/harness/deerflow/constants.py b/backend/packages/harness/deerflow/constants.py new file mode 100644 index 000000000..186675ac0 --- /dev/null +++ b/backend/packages/harness/deerflow/constants.py @@ -0,0 +1,3 @@ +"""Shared runtime protocol constants.""" + +DEFAULT_SKILLS_CONTAINER_PATH = "/mnt/skills" diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index d5b62c334..73de8e859 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -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 diff --git a/backend/packages/harness/deerflow/skills/slash.py b/backend/packages/harness/deerflow/skills/slash.py index f4dab54c3..c2540f6de 100644 --- a/backend/packages/harness/deerflow/skills/slash.py +++ b/backend/packages/harness/deerflow/skills/slash.py @@ -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) diff --git a/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py b/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py index f7e2959d5..cbb5f5c14 100644 --- a/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py +++ b/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py @@ -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 diff --git a/backend/packages/harness/deerflow/skills/storage/skill_storage.py b/backend/packages/harness/deerflow/skills/storage/skill_storage.py index 15af90102..7ebb0987e 100644 --- a/backend/packages/harness/deerflow/skills/storage/skill_storage.py +++ b/backend/packages/harness/deerflow/skills/storage/skill_storage.py @@ -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 # ------------------------------------------------------------------ diff --git a/backend/packages/harness/deerflow/skills/types.py b/backend/packages/harness/deerflow/skills/types.py index 2862828e3..4cb3ade74 100644 --- a/backend/packages/harness/deerflow/skills/types.py +++ b/backend/packages/harness/deerflow/skills/types.py @@ -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. diff --git a/backend/packages/harness/deerflow/subagents/__init__.py b/backend/packages/harness/deerflow/subagents/__init__.py index bd78d9acc..29f30a004 100644 --- a/backend/packages/harness/deerflow/subagents/__init__.py +++ b/backend/packages/harness/deerflow/subagents/__init__.py @@ -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}") diff --git a/backend/packages/harness/deerflow/subagents/status_contract.py b/backend/packages/harness/deerflow/subagents/status_contract.py index 03fa010af..6651dd296 100644 --- a/backend/packages/harness/deerflow/subagents/status_contract.py +++ b/backend/packages/harness/deerflow/subagents/status_contract.py @@ -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 diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 4940ff08b..88bac1d99 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -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) diff --git a/backend/tests/test_create_deerflow_agent.py b/backend/tests/test_create_deerflow_agent.py index 73ec09a48..dea50f1be 100644 --- a/backend/tests/test_create_deerflow_agent.py +++ b/backend/tests/test_create_deerflow_agent.py @@ -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 # --------------------------------------------------------------------------- diff --git a/backend/tests/test_delegation_ledger.py b/backend/tests/test_delegation_ledger.py index e14748734..28eed63de 100644 --- a/backend/tests/test_delegation_ledger.py +++ b/backend/tests/test_delegation_ledger.py @@ -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) diff --git a/backend/tests/test_durable_context_middleware.py b/backend/tests/test_durable_context_middleware.py index ea46537e6..d2554d225 100644 --- a/backend/tests/test_durable_context_middleware.py +++ b/backend/tests/test_durable_context_middleware.py @@ -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), ], diff --git a/backend/tests/test_gateway_imports.py b/backend/tests/test_gateway_imports.py new file mode 100644 index 000000000..09772ce67 --- /dev/null +++ b/backend/tests/test_gateway_imports.py @@ -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 diff --git a/backend/tests/test_skill_container_path_defaults.py b/backend/tests/test_skill_container_path_defaults.py new file mode 100644 index 000000000..5a61d7b97 --- /dev/null +++ b/backend/tests/test_skill_container_path_defaults.py @@ -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 == [] diff --git a/backend/tests/test_skill_context.py b/backend/tests/test_skill_context.py index 7088ca099..2564547b8 100644 --- a/backend/tests/test_skill_context.py +++ b/backend/tests/test_skill_context.py @@ -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) diff --git a/backend/tests/test_subagent_status_contract.py b/backend/tests/test_subagent_status_contract.py index 95e35ca42..886d30a38 100644 --- a/backend/tests/test_subagent_status_contract.py +++ b/backend/tests/test_subagent_status_contract.py @@ -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] diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index 4fc60e804..fef64e4d3 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -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. diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py index e30a47c97..922cf7cc8 100644 --- a/backend/tests/test_tool_error_handling_middleware.py +++ b/backend/tests/test_tool_error_handling_middleware.py @@ -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) diff --git a/backend/tests/test_tool_error_handling_subagent_stamp.py b/backend/tests/test_tool_error_handling_subagent_stamp.py index b130e5fe5..d6a748b58 100644 --- a/backend/tests/test_tool_error_handling_subagent_stamp.py +++ b/backend/tests/test_tool_error_handling_subagent_stamp.py @@ -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", diff --git a/contracts/subagent_status_contract.json b/contracts/subagent_status_contract.json index b89e1fdc1..6d998759d 100644 --- a/contracts/subagent_status_contract.json +++ b/contracts/subagent_status_contract.json @@ -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"] } diff --git a/frontend/src/content/en/harness/middlewares.mdx b/frontend/src/content/en/harness/middlewares.mdx index 0bdd05196..5f65f9c2c 100644 --- a/frontend/src/content/en/harness/middlewares.mdx +++ b/frontend/src/content/en/harness/middlewares.mdx @@ -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. --- diff --git a/frontend/src/content/zh/harness/middlewares.mdx b/frontend/src/content/zh/harness/middlewares.mdx index f7c123dc7..361bd9e92 100644 --- a/frontend/src/content/zh/harness/middlewares.mdx +++ b/frontend/src/content/zh/harness/middlewares.mdx @@ -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 技能引用捕获。 --- diff --git a/frontend/src/core/tasks/subtask-result.ts b/frontend/src/core/tasks/subtask-result.ts index 9efa45e8a..02143b69d 100644 --- a/frontend/src/core/tasks/subtask-result.ts +++ b/frontend/src/core/tasks/subtask-result.ts @@ -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 = { 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 | 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 | null | undefined, +): boolean { + if (!additionalKwargs) return false; + return STRUCTURED_SUBAGENT_KEYS.some((key) => + Object.prototype.hasOwnProperty.call(additionalKwargs, key), + ); +} + +function readStructuredResultBrief( + additionalKwargs: Record | null | undefined, +): string | undefined { + const value = additionalKwargs?.[SUBAGENT_RESULT_BRIEF_KEY]; + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} diff --git a/frontend/tests/unit/core/tasks/subtask-result.test.ts b/frontend/tests/unit/core/tasks/subtask-result.test.ts index 1d676b49c..f90add980 100644 --- a/frontend/tests/unit/core/tasks/subtask-result.test.ts +++ b/frontend/tests/unit/core/tasks/subtask-result.test.ts @@ -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)); }); } });