diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index c8daf35de..54a3427d2 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/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): diff --git a/backend/packages/harness/deerflow/agents/factory.py b/backend/packages/harness/deerflow/agents/factory.py index df2c64894..35b04ee1c 100644 --- a/backend/packages/harness/deerflow/agents/factory.py +++ b/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()) diff --git a/backend/tests/test_create_deerflow_agent.py b/backend/tests/test_create_deerflow_agent.py index 55c27c1ea..c7c29d83b 100644 --- a/backend/tests/test_create_deerflow_agent.py +++ b/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"