fix(agents): make create_deerflow_agent's subagent limit, summarization and token_budget features take effect (#5488)

* fix(agents): keep the delegation ledger and summary in create_deerflow_agent graphs

The SDK factory chain had no DurableContextMiddleware. SubagentLimitMiddleware
counts a run's delegations from the ledger that middleware writes, so the
per-run subagent total never tripped, and DeerFlowSummarizationMiddleware
keeps compacted history in summary_text, which only that middleware puts back
into model requests, so a summarized factory graph lost its history.

Add it after ToolErrorHandlingMiddleware, ahead of summarization, where
make_lead_agent has it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(agents): make RuntimeFeatures(token_budget=True) enforce the budget

The factory built TokenBudgetMiddleware from TokenBudgetConfig(), whose
enabled flag defaults to False, and every hook returns early on it. A graph
created with token_budget=True had no warning, no hard stop and no
token_capped.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(agents): coalesce system messages in create_deerflow_agent graphs

DurableContextMiddleware injects its authority contract as a second
SystemMessage. The lead and subagent chains pair it with
SystemMessageCoalescingMiddleware because strict backends reject that; the
factory now does the same.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* docs: trim inherited harness guidance below chain limits

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/AGENTS.md

@@ -52,21 +52,21 @@ drift.
 
 ### Embedded Client (`packages/harness/deerflow/client.py`)
 
-`DeerFlowClient` provides in-process access without HTTP or a FastAPI dependency. It shares Gateway's `deerflow` modules, config files, data directories, and response schemas for compatible consumers.
+`DeerFlowClient` provides in-process access without HTTP/FastAPI, sharing Gateway's `deerflow` modules, config, data directories, and response schemas.
 
 **Agent Conversation**:
 - `chat(message, thread_id)` — synchronous, accumulates streaming deltas per message-id and returns the final AI text
 - `stream(message, thread_id)` — subscribes to LangGraph `stream_mode=["values", "messages", "custom"]` and yields `StreamEvent`:
-  - `"values"` — state snapshot (title, messages, artifacts, summary_text); `summary_text` is the current summary or `None` when absent and is forwarded on every snapshot, including unchanged summaries and resets. AI text already delivered via `messages` mode is **not** re-synthesized here to avoid duplicate deliveries; serialized `ToolMessage` entries preserve a non-`None` native `artifact`
-  - `"messages-tuple"` — per-chunk update: for AI text this is a **delta** (concat per `id` to rebuild the full message); tool calls and tool results are emitted once each, and tool results preserve a non-`None` native `artifact`
+  - `"values"` — state snapshot (title, messages, artifacts, summary_text). Always forward `summary_text` (current summary or `None`), including unchanged values/resets. Never re-emit AI text delivered via `messages`; serialized `ToolMessage` entries retain non-`None` native `artifact`
+  - `"messages-tuple"` — AI text **deltas** (concatenate per `id`); emit tool calls/results once each, preserving non-`None` native result `artifact`
   - `"custom"` — forwarded from `StreamWriter`; DeerFlow-built-in custom events are dual-emitted through `deerflow.utils.custom_events`, so `astream_events(version="v2")` consumers also receive one `on_custom_event` with `name=payload["type"]` and the unchanged payload as `data`
   - `"end"` — stream finished (carries cumulative `usage` counted once per message id)
-- **Custom-event invariant** — production DeerFlow emitters must use `emit_custom_event` / `aemit_custom_event`, not call `StreamWriter` alone. Every built-in payload must carry a non-empty string `type`; typeless payloads remain writer-only and are intentionally absent from `astream_events`. The writer runs first and remains authoritative for Gateway, Web UI, and embedded-client compatibility; callback dispatch is best-effort and must not break that path. Async graph hooks must await the async helper rather than invoking synchronous dispatch on a running event loop.
+- **Custom-event invariant** — use `emit_custom_event` / `aemit_custom_event`, never `StreamWriter` alone. Built-in payloads require a non-empty string `type`; typeless payloads stay writer-only, absent from `astream_events`. The writer runs first and is authoritative for Gateway/Web UI/embedded clients; best-effort callbacks must not break it. Async graph hooks must await the async helper, never dispatch synchronously on a running event loop.
 - Agent created lazily via `create_agent()` + `build_middlewares()`, same as `make_lead_agent`
 - Cache graphs by effective storage `user_id` in every auth mode because prompts and middleware bind user SOUL, skills, and storage. `stream()` must materialize it before worker or isolated-loop boundaries.
 - Supports `checkpointer` parameter for state persistence across turns
 - `reset_agent()` forces agent recreation (e.g. after memory or skill changes)
-- See [docs/STREAMING.md](../../../docs/STREAMING.md) for the full design: why Gateway and DeerFlowClient are parallel paths, LangGraph's `stream_mode` semantics, the per-id dedup invariants, and regression testing strategy
+- [Streaming design](../../../docs/STREAMING.md): Gateway/client parallel paths, LangGraph `stream_mode`, per-id deduplication, and regression tests
 
 **Gateway Equivalent Methods** (replaces Gateway API):
 

backend/packages/harness/deerflow/agents/factory.py

@@ -210,6 +210,8 @@ def _assemble_from_features(
       3.   DanglingToolCallMiddleware (always)
       4.   GuardrailMiddleware (guardrail feature)
       5.   ToolErrorHandlingMiddleware (always)
+      5a.  DurableContextMiddleware (always)
+      5b.  SystemMessageCoalescingMiddleware (always)
       6.   SummarizationMiddleware (summarization feature)
       7.   TodoMiddleware (plan_mode parameter)
       8.   TitleMiddleware (auto_title feature)
@@ -258,6 +260,20 @@ def _assemble_from_features(
     # --- [5] ToolErrorHandling (always) ---
     chain.append(ToolErrorHandlingMiddleware())
 
+    # --- [5a] DurableContext (always) ---
+    # Summarization moves compacted history into ``summary_text``, and
+    # SubagentLimitMiddleware counts the run's delegations from the
+    # ``delegations`` ledger. This middleware writes that ledger and projects
+    # both into model requests. It sits ahead of summarization, as in
+    # make_lead_agent, so delegations are captured before they are compacted.
+    from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
+    from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware
+
+    chain.append(DurableContextMiddleware())
+    # DurableContext adds its authority contract as a second SystemMessage; strict backends
+    # (vLLM, SGLang, Qwen, Anthropic) reject that, so merge them into one leading message.
+    chain.append(SystemMessageCoalescingMiddleware())
+
     # --- [6] Summarization ---
     if feat.summarization is not False:
         if isinstance(feat.summarization, AgentMiddleware):
@@ -390,7 +406,8 @@ def _assemble_from_features(
             from deerflow.agents.middlewares.token_budget_middleware import TokenBudgetMiddleware
             from deerflow.config.token_budget_config import TokenBudgetConfig
 
-            chain.append(TokenBudgetMiddleware.from_config(TokenBudgetConfig()))
+            # ``enabled`` defaults to False for config.yaml; ``token_budget=True`` is the opt-in.
+            chain.append(TokenBudgetMiddleware.from_config(TokenBudgetConfig(enabled=True)))
 
     # --- [14] Clarification (always last among built-ins) ---
     chain.append(ClarificationMiddleware())

backend/tests/test_create_deerflow_agent.py

@@ -6,12 +6,14 @@ from unittest.mock import MagicMock, patch
 import pytest
 from langchain.agents import AgentState
 from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
-from langchain_core.messages import AIMessage
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
+from langchain_core.tools import tool
 from langgraph.channels import DeltaChannel
 from langgraph.checkpoint.memory import InMemorySaver
 
 from deerflow.agents.factory import create_deerflow_agent
 from deerflow.agents.features import Next, Prev, RuntimeFeatures
+from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware
 from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
 from deerflow.agents.thread_state import DeltaThreadState, ThreadState
 from deerflow.config.subagent_batches_config import SubagentBatchesConfig
@@ -932,6 +934,8 @@ def test_full_chain_order(mock_create_agent):
         "DanglingToolCallMiddleware",
         "MyGuardrail",
         "ToolErrorHandlingMiddleware",
+        "DurableContextMiddleware",
+        "SystemMessageCoalescingMiddleware",
         "MySummarization",
         "TodoMiddleware",
         "TitleMiddleware",
@@ -1057,3 +1061,109 @@ def test_extra_circular_dependency():
             features=RuntimeFeatures(sandbox=False),
             extra_middleware=[MW_A(), MW_B()],
         )
+
+
+# ===========================================================================
+# Delegation ledger and compacted summary in a factory-built graph
+# ===========================================================================
+
+
+class _RecordingFakeModel(_FakeModel):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        object.__setattr__(self, "received", [])
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        self.received.append(list(messages))
+        return super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+@tool("task")
+def _fake_task(description: str, prompt: str, subagent_type: str) -> str:
+    """Fake task tool."""
+    return f"Task Succeeded. Result: {description}"
+
+
+def _task_turn(call_id: str) -> AIMessage:
+    return AIMessage(
+        content="",
+        tool_calls=[{"name": "task", "args": {"description": call_id, "prompt": "do it", "subagent_type": "general-purpose"}, "id": call_id, "type": "tool_call"}],
+    )
+
+
+# ---------------------------------------------------------------------------
+# 43. The per-run subagent total holds across model turns
+# ---------------------------------------------------------------------------
+def test_subagent_total_per_run_holds_across_model_turns():
+    model = _FakeModel(responses=[_task_turn("call-1"), _task_turn("call-2"), _task_turn("call-3"), AIMessage(content="done")])
+    runtime = SubagentRuntime(SubagentRuntimeConfig(max_running=3), max_total_per_run=2)
+    graph = create_deerflow_agent(
+        model,
+        tools=[_fake_task],
+        features=RuntimeFeatures(subagent=True, sandbox=False),
+        subagent_runtime=runtime,
+    )
+
+    result = graph.invoke({"messages": [HumanMessage(content="delegate three pieces of work")]}, context={"run_id": "run-1"})
+
+    ran = [message.tool_call_id for message in result["messages"] if isinstance(message, ToolMessage) and message.name == "task"]
+    assert ran == ["call-1", "call-2"]
+
+
+# ---------------------------------------------------------------------------
+# 44. The model still sees the summary after summarization compacts history
+# ---------------------------------------------------------------------------
+def test_summarization_feature_keeps_the_summary_in_model_requests():
+    summarizer = DeerFlowSummarizationMiddleware(
+        model=_FakeModel(responses=[AIMessage(content="compressed summary")]),
+        trigger=("messages", 4),
+        keep=("messages", 2),
+        token_counter=len,
+    )
+    model = _RecordingFakeModel(responses=[AIMessage(content="one"), AIMessage(content="two"), AIMessage(content="three")])
+    graph = create_deerflow_agent(
+        model,
+        system_prompt="You are a test agent.",
+        features=RuntimeFeatures(summarization=summarizer, sandbox=False),
+        checkpointer=InMemorySaver(),
+    )
+    config = {"configurable": {"thread_id": "factory-summary"}}
+
+    for text in ("first", "second", "third"):
+        result = graph.invoke({"messages": [HumanMessage(content=text)]}, config)
+
+    assert result["summary_text"] == "compressed summary"
+    assert "first" not in [message.content for message in result["messages"]]
+    request = model.received[-1]
+    assert any("compressed summary" in str(message.content) for message in request)
+    # The durable-context authority contract must not reach the provider as a second SystemMessage.
+    assert [index for index, message in enumerate(request) if isinstance(message, SystemMessage)] == [0]
+    assert "You are a test agent." in request[0].content
+
+
+# ---------------------------------------------------------------------------
+# 45. token_budget=True enforces the default budget
+# ---------------------------------------------------------------------------
+def test_token_budget_true_enforces_the_default_budget():
+    @tool("bash")
+    def bash(command: str) -> str:
+        """Run a fake shell command."""
+        return "ok"
+
+    over_budget = AIMessage(
+        content="",
+        tool_calls=[{"name": "bash", "args": {"command": "ls"}, "id": "call-1", "type": "tool_call"}],
+        usage_metadata={"input_tokens": 250_000, "output_tokens": 0, "total_tokens": 250_000},
+    )
+    graph = create_deerflow_agent(
+        _FakeModel(responses=[over_budget, AIMessage(content="done")]),
+        tools=[bash],
+        features=RuntimeFeatures(token_budget=True, sandbox=False),
+    )
+    context = {"run_id": "run-1"}
+
+    result = graph.invoke({"messages": [HumanMessage(content="go")]}, context=context)
+
+    assert not any(isinstance(message, ToolMessage) for message in result["messages"])
+    assert "TOKEN BUDGET EXCEEDED" in result["messages"][-1].content
+    assert context["stop_reason"] == "token_capped"