diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 068de2ef7..59d88e299 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -190,8 +190,8 @@ from deerflow.config import get_app_config - System prompt generated by `apply_prompt_template()` with skills, memory, and subagent instructions **ThreadState** (`packages/harness/deerflow/agents/thread_state.py`): -- Extends `AgentState` with: `sandbox`, `thread_data`, `title`, `artifacts`, `todos`, `uploaded_files`, `viewed_images` -- Uses custom reducers: `merge_artifacts` (deduplicate), `merge_viewed_images` (merge/clear) +- Extends `AgentState` with: `sandbox`, `thread_data`, `title`, `artifacts`, `todos`, `uploaded_files`, `viewed_images`, `delegations` +- Uses custom reducers: `merge_artifacts` (deduplicate), `merge_viewed_images` (merge/clear), `merge_delegations` (upsert subagent delegation ledger by `task_id`; terminal status never downgraded) **Runtime Configuration** (via `config.configurable`): - `thinking_enabled` - Enable model's extended thinking @@ -227,13 +227,14 @@ Lead-agent middlewares are assembled in strict order across three functions: the 17. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses) 18. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call 19. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) -20. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives -21. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit -22. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer -23. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits -24. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail -25. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first -26. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) +20. **DelegationLedgerMiddleware** - *(optional, if `subagent_enabled`)* Maintains a system-maintained ledger of delegated subtasks in `ThreadState.delegations` (derives `description` + status from `task` tool calls and their result `subagent_status` on `after_model`) and re-injects it as a hidden `` on `wrap_model_call` so the lead stops re-delegating the same work. Stored in state (not message history) so it survives summarization; re-injected ephemerally each call so no duplicate snapshots accumulate. Registered before SystemMessageCoalescingMiddleware so its injected SystemMessage is folded into the leading one +21. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives +22. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit +23. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer +24. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits +25. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail +26. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first +27. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) ### Configuration System diff --git a/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/backend/packages/harness/deerflow/agents/lead_agent/agent.py index 1beff6337..4b43b0be2 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -348,6 +348,14 @@ def build_middlewares( middlewares.append(DeferredToolFilterMiddleware(deferred_setup.deferred_names, deferred_setup.catalog_hash)) + # Maintain + inject the subagent delegation ledger (only when delegation is on). + # Registered before coalescing so its injected SystemMessage + # is folded into the single leading SystemMessage for strict backends. + if cfg.get("subagent_enabled", False): + from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware + + middlewares.append(DelegationLedgerMiddleware()) + # Coalesce every SystemMessage into a single leading one before the request # reaches the provider. Strict backends (vLLM, SGLang, Qwen, Anthropic) # reject non-leading SystemMessages. See system_message_coalescing_middleware.py. diff --git a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger_middleware.py new file mode 100644 index 000000000..0ec48f315 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger_middleware.py @@ -0,0 +1,109 @@ +"""Lead-agent middleware: a system-maintained ledger of delegated subtasks. + +Issue: the lead repeatedly re-delegated the same research because the context +held no durable record of what it had already dispatched (the record was lost +to summarization). This middleware keeps that record in ThreadState (which +summarization does not touch) and re-injects it into every model call, so the +model always sees "already delegated: ..." and stops re-delegating. +""" + +from __future__ import annotations + +import logging +from collections.abc import Awaitable, Callable +from typing import Any + +from langchain.agents.middleware import AgentMiddleware, ModelRequest +from langchain_core.messages import AIMessage, SystemMessage, ToolMessage +from langgraph.runtime import Runtime + +from deerflow.agents.thread_state import DelegationEntry +from deerflow.subagents.status_contract import SUBAGENT_STATUS_KEY, extract_subagent_status + +logger = logging.getLogger(__name__) + + +def extract_delegations(messages: list) -> list[DelegationEntry]: + """Derive delegation entries from the visible message list, in dispatch order. + + A ``task`` tool-call is a dispatch (status "in_progress"); its matching + ToolMessage upgrades the status from the structured ``subagent_status`` kwarg, + or by parsing the result text as a fallback (same contract the frontend uses). + """ + by_id: dict[str, DelegationEntry] = {} + + for message in messages: + if isinstance(message, AIMessage): + for call in message.tool_calls or []: + if call.get("name") != "task": + continue + task_id = call.get("id") + if not task_id or task_id in by_id: + continue + args = call.get("args") or {} + by_id[task_id] = { + "task_id": task_id, + "description": args.get("description") or "", + "subagent_type": args.get("subagent_type") or "", + "status": "in_progress", + } + elif isinstance(message, ToolMessage): + entry = by_id.get(message.tool_call_id or "") + if entry is None: + continue + status = message.additional_kwargs.get(SUBAGENT_STATUS_KEY) + if not status: + content = message.content if isinstance(message.content, str) else str(message.content) + status = extract_subagent_status(content) + if status: + entry["status"] = status + + return list(by_id.values()) + + +def format_delegation_block(entries: list[DelegationEntry]) -> str | None: + """Render the ledger as a hidden , or None when empty.""" + if not entries: + return None + + lines = [ + "", + "", + "You have ALREADY delegated these subtasks in this run. Do NOT re-delegate the same work; reuse or build on their results instead.", + ] + for entry in entries: + lines.append(f"- [{entry['status']}] ({entry['subagent_type']}) {entry['description']}") + lines.append("") + lines.append("") + return "\n".join(lines) + + +class DelegationLedgerMiddleware(AgentMiddleware): + """Maintain (after_model) and inject (wrap_model_call) the delegation ledger.""" + + def _derive_update(self, state: Any) -> dict | None: + entries = extract_delegations(list(state.get("messages", []))) + return {"delegations": entries} if entries else None + + def after_model(self, state: Any, runtime: Runtime | None = None) -> dict | None: + return self._derive_update(state) + + async def aafter_model(self, state: Any, runtime: Runtime | None = None) -> dict | None: + return self._derive_update(state) + + def _inject(self, request: ModelRequest) -> ModelRequest: + entries = list(request.state.get("delegations") or []) + block = format_delegation_block(entries) + + if not block: + logger.debug("delegation ledger: nothing to inject this call") + return request + logger.info("delegation ledger: injected %d subtask(s) into the model request", len(entries)) + reminder = SystemMessage(content=block, additional_kwargs={"hide_from_ui": True}) + return request.override(messages=[*request.messages, reminder]) + + def wrap_model_call(self, request: ModelRequest, handler: Callable[[ModelRequest], Any]) -> Any: + return handler(self._inject(request)) + + async def awrap_model_call(self, request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[Any]]) -> Any: + return await handler(self._inject(request)) diff --git a/backend/packages/harness/deerflow/agents/thread_state.py b/backend/packages/harness/deerflow/agents/thread_state.py index b88b58129..6fb6a3476 100644 --- a/backend/packages/harness/deerflow/agents/thread_state.py +++ b/backend/packages/harness/deerflow/agents/thread_state.py @@ -2,6 +2,8 @@ from typing import Annotated, NotRequired, TypedDict from langchain.agents import AgentState +from deerflow.subagents.status_contract import SUBAGENT_STATUS_VALUES + class SandboxState(TypedDict): sandbox_id: NotRequired[str | None] @@ -108,6 +110,42 @@ def merge_promoted(existing: PromotedTools | None, new: PromotedTools | None) -> } +# Terminal subagent statuses. Derived from the single source of truth +# (SUBAGENT_STATUS_VALUES) so the set can never drift from the status contract: +# every value the contract enumerates is terminal, and the only non-terminal +# status, "in_progress", is intentionally absent from the contract. merge_delegations +# uses this to guard against status downgrades. test_delegation_ledger pins the +# derivation so a future contract edit cannot silently desync this set. +TERMINAL_STATUSES: frozenset[str] = frozenset(SUBAGENT_STATUS_VALUES) + + +class DelegationEntry(TypedDict): + task_id: str + description: str + subagent_type: str + status: str # "in_progress" or one of TERMINAL_STATUSES + + +def merge_delegations( + existing: list[DelegationEntry] | None, + new: list[DelegationEntry] | None, +) -> list[DelegationEntry]: + """Reducer for the delegation ledger: upsert by task_id, preserve dispatch order. + + A terminal status is never overwritten by a non-terminal one, so a later + re-derivation from a partially-summarized message list cannot regress a + finished subtask back to "in_progress". + """ + merged: dict[str, DelegationEntry] = {} + for entry in list(existing or []) + list(new or []): + task_id = entry["task_id"] + prev = merged.get(task_id) + if prev is not None and prev["status"] in TERMINAL_STATUSES and entry["status"] not in TERMINAL_STATUSES: + continue + merged[task_id] = {**prev, **entry} if prev else dict(entry) + return list(merged.values()) + + class ThreadState(AgentState): sandbox: SandboxStateField thread_data: NotRequired[ThreadDataState | None] @@ -117,3 +155,4 @@ class ThreadState(AgentState): uploaded_files: NotRequired[list[dict] | None] viewed_images: Annotated[dict[str, ViewedImageData], merge_viewed_images] # image_path -> {base64, mime_type} promoted: Annotated[PromotedTools | None, merge_promoted] + delegations: Annotated[list[DelegationEntry], merge_delegations] diff --git a/backend/tests/fixtures/replay/write_read_file.ultra.events.json b/backend/tests/fixtures/replay/write_read_file.ultra.events.json index 83976df10..9b6947f51 100644 --- a/backend/tests/fixtures/replay/write_read_file.ultra.events.json +++ b/backend/tests/fixtures/replay/write_read_file.ultra.events.json @@ -13,6 +13,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "viewed_images" ] @@ -21,6 +22,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "thread_data", "viewed_images" @@ -30,6 +32,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "thread_data", "viewed_images" @@ -39,6 +42,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "thread_data", "viewed_images" @@ -48,6 +52,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "thread_data", "title", @@ -58,6 +63,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "thread_data", "title", @@ -68,6 +74,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "sandbox", "thread_data", @@ -79,6 +86,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "sandbox", "thread_data", @@ -90,6 +98,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "sandbox", "thread_data", @@ -101,6 +110,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "sandbox", "thread_data", @@ -112,6 +122,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "sandbox", "thread_data", @@ -123,6 +134,7 @@ "event": "values", "keys": [ "artifacts", + "delegations", "messages", "sandbox", "thread_data", diff --git a/backend/tests/test_delegation_ledger.py b/backend/tests/test_delegation_ledger.py new file mode 100644 index 000000000..5655b7751 --- /dev/null +++ b/backend/tests/test_delegation_ledger.py @@ -0,0 +1,234 @@ +"""Tests for the subagent delegation ledger (parent issue: redundant delegation). + +The ledger is a system-maintained record of "subtasks already delegated + their +status", stored in ThreadState (so it survives summarization) and re-injected +into context each model call so the lead stops re-delegating the same work. +""" + +from langchain_core.messages import AIMessage, ToolMessage + +from deerflow.agents.middlewares.delegation_ledger_middleware import ( + extract_delegations, + format_delegation_block, +) +from deerflow.agents.thread_state import TERMINAL_STATUSES, merge_delegations +from deerflow.subagents.status_contract import SUBAGENT_STATUS_VALUES + + +def _entry(task_id, status, description="d", subagent_type="general-purpose"): + return {"task_id": task_id, "description": description, "subagent_type": subagent_type, "status": status} + + +def _task_call(task_id, description, subagent_type="general-purpose"): + return {"name": "task", "args": {"description": description, "subagent_type": subagent_type}, "id": task_id, "type": "tool_call"} + + +def test_terminal_statuses_derived_from_status_contract(): + """TERMINAL_STATUSES must stay the exact set the status contract enumerates. + + Pins the derivation in thread_state.py: every value the contract declares is a + terminal status, and the lone non-terminal status "in_progress" is never part of + the contract. If a future contract edit adds a non-terminal value (or otherwise + changes the set), this fails loudly instead of letting merge_delegations' + downgrade guard silently desync. + """ + assert TERMINAL_STATUSES == frozenset(SUBAGENT_STATUS_VALUES) + assert "in_progress" not in TERMINAL_STATUSES + + +def test_merge_upserts_by_task_id_preserving_order(): + existing = [_entry("a", "in_progress"), _entry("b", "in_progress")] + new = [_entry("b", "completed"), _entry("c", "in_progress")] + + merged = merge_delegations(existing, new) + + assert [e["task_id"] for e in merged] == ["a", "b", "c"] + assert next(e for e in merged if e["task_id"] == "b")["status"] == "completed" + + +def test_merge_does_not_downgrade_terminal_status(): + existing = [_entry("a", "completed")] + new = [_entry("a", "in_progress")] + + merged = merge_delegations(existing, new) + + assert merged[0]["status"] == "completed" + + +def test_merge_handles_none_inputs(): + assert merge_delegations(None, None) == [] + assert merge_delegations(None, [_entry("a", "in_progress")])[0]["task_id"] == "a" + assert merge_delegations([_entry("a", "in_progress")], None)[0]["task_id"] == "a" + + +def test_extract_records_dispatch_as_in_progress(): + msgs = [AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")])] + + entries = extract_delegations(msgs) + + assert entries == [{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "in_progress"}] + + +def test_extract_updates_status_from_tool_message_kwarg(): + msgs = [ + AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")]), + ToolMessage(content="Task Succeeded. Result: ok", tool_call_id="call_1", additional_kwargs={"subagent_status": "completed"}), + ] + + entries = extract_delegations(msgs) + + assert entries[0]["status"] == "completed" + + +def test_extract_falls_back_to_parsing_content_when_kwarg_absent(): + msgs = [ + AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")]), + ToolMessage(content="Task failed. Error: boom", tool_call_id="call_1"), + ] + + entries = extract_delegations(msgs) + + assert entries[0]["status"] == "failed" + + +def test_extract_ignores_non_task_tool_calls(): + msgs = [AIMessage(content="", tool_calls=[{"name": "web_search", "args": {}, "id": "x", "type": "tool_call"}])] + + assert extract_delegations(msgs) == [] + + +def test_extract_preserves_dispatch_order_across_batches(): + msgs = [ + AIMessage(content="", tool_calls=[_task_call("call_1", "A"), _task_call("call_2", "B")]), + AIMessage(content="", tool_calls=[_task_call("call_3", "C")]), + ] + + assert [e["task_id"] for e in extract_delegations(msgs)] == ["call_1", "call_2", "call_3"] + + +def test_format_block_lists_entries_and_returns_none_when_empty(): + assert format_delegation_block([]) is None + + block = format_delegation_block( + [ + {"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "completed"}, + {"task_id": "call_2", "description": "Research B", "subagent_type": "general-purpose", "status": "in_progress"}, + ] + ) + + assert "" in block + assert "Research A" in block and "completed" in block + assert "Research B" in block and "in_progress" in block + assert "re-delegate" in block.lower() or "already delegated" in block.lower() + + +def test_after_model_returns_derived_delegations(): + from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware + + mw = DelegationLedgerMiddleware() + state = {"messages": [AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")])]} + + update = mw.after_model(state, runtime=None) + + assert update == {"delegations": [{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "in_progress"}]} + + +def test_after_model_returns_none_when_no_delegations(): + from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware + + mw = DelegationLedgerMiddleware() + state = {"messages": [AIMessage(content="hi")]} + + assert mw.after_model(state, runtime=None) is None + + +class _FakeRequest: + """Minimal stand-in for ModelRequest: holds state + messages, supports override().""" + + def __init__(self, state, messages): + self.state = state + self.messages = messages + + def override(self, *, messages): + return _FakeRequest(self.state, messages) + + +def test_wrap_model_call_injects_ledger_block(): + from langchain_core.messages import SystemMessage + + from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware + + mw = DelegationLedgerMiddleware() + captured = {} + + def handler(req): + captured["messages"] = req.messages + return "RESPONSE" + + state = {"delegations": [{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "completed"}]} + req = _FakeRequest(state, [AIMessage(content="prev")]) + + result = mw.wrap_model_call(req, handler) + + assert result == "RESPONSE" + injected = captured["messages"] + assert isinstance(injected[-1], SystemMessage) + assert "Research A" in injected[-1].content + assert len(injected) == 2 + + +def test_wrap_model_call_is_noop_without_delegations(): + from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware + + mw = DelegationLedgerMiddleware() + captured = {} + + def handler(req): + captured["messages"] = req.messages + return "RESPONSE" + + req = _FakeRequest({"delegations": []}, [AIMessage(content="prev")]) + + mw.wrap_model_call(req, handler) + + assert len(captured["messages"]) == 1 + + +def _mw_names(middlewares): + return [type(m).__name__ for m in middlewares] + + +def _explicit_app_config(): + """Build a minimal in-memory AppConfig (with one model) so build_middlewares + never reads the gitignored, CI-absent config.yaml via get_app_config().""" + from deerflow.config.app_config import AppConfig + from deerflow.config.model_config import ModelConfig + from deerflow.config.sandbox_config import SandboxConfig + + model = ModelConfig( + name="test-model", + display_name="test-model", + description=None, + use="langchain_openai:ChatOpenAI", + model="test-model", + supports_thinking=False, + supports_vision=False, + ) + return AppConfig(models=[model], sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider")) + + +def test_middleware_registered_when_subagent_enabled(): + from deerflow.agents.lead_agent.agent import build_middlewares + + middlewares = build_middlewares({"configurable": {"subagent_enabled": True}}, None, app_config=_explicit_app_config()) + names = _mw_names(middlewares) + assert "DelegationLedgerMiddleware" in names + # Must run before coalescing so its injected SystemMessage gets folded in. + assert names.index("DelegationLedgerMiddleware") < names.index("SystemMessageCoalescingMiddleware") + + +def test_middleware_absent_when_subagent_disabled(): + from deerflow.agents.lead_agent.agent import build_middlewares + + middlewares = build_middlewares({"configurable": {"subagent_enabled": False}}, None, app_config=_explicit_app_config()) + assert "DelegationLedgerMiddleware" not in _mw_names(middlewares)