[argus] todo_middleware: ArgusTodoMiddleware variant for planner-driven agents

Adds a TodoMiddleware subclass whose default system prompt and tool
description defer planning judgment to a skill (the planner SKILL.md
in the consumer's deployment) rather than asking the agent to make
its own "is this complex enough?" call. The factory selects the
subclass when the runtime context has agent_name=="qwen-local-coder";
all other agents continue to receive the upstream TodoMiddleware
unchanged.

Why: a planner-skill-driven agent that mandates "always plan, planner
decides skip vs real plan" runs into a direct contradiction with the
upstream TodoList prompt's "DO NOT use this tool for simple tasks
(< 3 steps)" guidance. With both active the agent follows the
middleware prompt and skips the planner. Reproduced empirically on
bare prompts (e.g. "Create a webpage that shows a clock") that the
planner skill is supposed to catch.

Subclass over inline branching because:
- one new file, one new class, single source of override
- future planner-pattern agents opt in by adding their name to the
  factory dispatch
- behavior (context-loss reminders in before_model, premature-exit
  prevention in after_model) inherits from TodoMiddleware unchanged

Files:
- backend/packages/harness/deerflow/agents/middlewares/argus_todo_middleware.py
  ArgusTodoMiddleware: same constructor signature as TodoMiddleware,
  Argus-aligned defaults for system_prompt and tool_description.
- backend/packages/harness/deerflow/agents/lead_agent/agent.py
  _create_todo_list_middleware now takes agent_name and routes
  qwen-local-coder onto ArgusTodoMiddleware. _build_middlewares
  forwards the existing agent_name argument it already had for
  MemoryMiddleware.
- backend/tests/test_argus_todo_middleware.py
  7 unit tests: construction defaults, kwarg overrides, inherited
  reminder behavior, factory routing for qwen-local-coder, factory
  routing for code-reviewer (regression check), factory with empty
  agent_name (default upstream), factory returns None when
  is_plan_mode is off.

Argus repo iteration 6.3 will pin this SHA and re-bind the
"Planner populates todos from steps" matcher to its eval tasks.

Author: Nichol4s

backend/packages/harness/deerflow/agents/lead_agent/agent.py

@@ -113,18 +113,34 @@ def _create_summarization_middleware() -> DeerFlowSummarizationMiddleware | None
     )
 
 
-def _create_todo_list_middleware(is_plan_mode: bool) -> TodoMiddleware | None:
+def _create_todo_list_middleware(is_plan_mode: bool, agent_name: str | None = "") -> TodoMiddleware | None:
     """Create and configure the TodoList middleware.
 
     Args:
         is_plan_mode: Whether to enable plan mode with TodoList middleware.
+        agent_name: Custom-agent name from the runtime context. When set to
+            ``"qwen-local-coder"``, returns ``ArgusTodoMiddleware`` whose
+            system prompt and tool description defer planning judgment to the
+            planner SKILL.md (Argus iteration 6.3). Any other value (or empty)
+            returns the upstream ``TodoMiddleware`` with the standard prompt.
 
     Returns:
-        TodoMiddleware instance if plan mode is enabled, None otherwise.
+        TodoMiddleware (or subclass) instance if plan mode is enabled, None otherwise.
     """
     if not is_plan_mode:
         return None
 
+    # Argus override: qwen-local-coder has a mandatory planner skill that
+    # decides when to plan and what the steps are. The upstream system prompt
+    # below tells the agent NOT to use write_todos for "simple tasks", which
+    # contradicts the planner's "always plan" mandate. ArgusTodoMiddleware
+    # replaces only the prompt; behavior (reminders, exit prevention)
+    # inherits unchanged.
+    if agent_name == "qwen-local-coder":
+        from deerflow.agents.middlewares.argus_todo_middleware import ArgusTodoMiddleware
+
+        return ArgusTodoMiddleware()
+
     # Custom prompts matching DeerFlow's style
     system_prompt = """
 <todo_list_system>
@@ -256,10 +272,12 @@ def _build_middlewares(config: RunnableConfig, model_name: str | None, agent_nam
     if summarization_middleware is not None:
         middlewares.append(summarization_middleware)
 
-    # Add TodoList middleware if plan mode is enabled
+    # Add TodoList middleware if plan mode is enabled. agent_name routes
+    # qwen-local-coder onto ArgusTodoMiddleware (planner-aligned prompt);
+    # other agents stay on the upstream TodoMiddleware.
     cfg = _get_runtime_config(config)
     is_plan_mode = cfg.get("is_plan_mode", False)
-    todo_list_middleware = _create_todo_list_middleware(is_plan_mode)
+    todo_list_middleware = _create_todo_list_middleware(is_plan_mode, agent_name=agent_name)
     if todo_list_middleware is not None:
         middlewares.append(todo_list_middleware)
 

backend/packages/harness/deerflow/agents/middlewares/argus_todo_middleware.py

@@ -0,0 +1,90 @@
+"""Argus variant of TodoMiddleware that defers planning judgment to the planner skill.
+
+Upstream's TodoMiddleware injects a system prompt and tool description telling
+the agent NOT to use ``write_todos`` for "simple tasks (< 3 steps)" and to make
+its own judgment about when planning is warranted. Argus's qwen-local-coder
+agent has a planner SKILL.md that decides when to plan and what the steps are;
+``write_todos`` is then used to mirror those steps into ``state.todos[]`` for
+live UI tick-off.
+
+The two systems contradict each other if both are active: the agent reads
+upstream's "skip this for simple tasks" guidance and skips the planner on
+bare prompts the planner SKILL.md would have caught. This subclass overrides
+only the prompt and tool-description text so the agent reads "the planner
+decides" instead. All other behavior — context-loss reminders in
+``before_model``, premature-exit prevention in ``after_model``, async hooks —
+inherits from the parent unchanged.
+
+Selected by ``deerflow.agents.lead_agent.agent._create_todo_list_middleware``
+when ``agent_name == "qwen-local-coder"``. Other agents continue to receive
+the upstream prompt.
+"""
+
+from __future__ import annotations
+
+from .todo_middleware import TodoMiddleware
+
+
+_ARGUS_SYSTEM_PROMPT = """
+<todo_list_system>
+You have access to the `write_todos` tool. It maintains a live progress view in the user's UI by writing the current step list into `state.todos[]`.
+
+The `planner` skill (read /mnt/skills/public/planner/SKILL.md) is the source of truth for when to plan and what the steps are. If the planner writes a real plan to /mnt/user-data/workspace/plan.json, mirror its `steps[]` into write_todos:
+
+- One todo per step, content = step.action.
+- First todo `status: "in_progress"` (about to start).
+- Subsequent todos `status: "pending"`.
+- Before starting step Si (i > 1): flip Si-1 to "completed" and Si to "in_progress" via another write_todos call.
+- After the last step: one final write_todos call marking it "completed".
+
+If the planner returns the skip form (`{"status": "skip", ...}`), do NOT call write_todos at all. The user gets a direct answer.
+
+The planner's decision supersedes any general guidance about "complex vs. simple tasks." If the planner produced a real plan, mirror it. If it skipped, don't.
+</todo_list_system>
+"""
+
+
+_ARGUS_TOOL_DESCRIPTION = """Mirror the planner's plan.json steps[] into the live state.todos[] for UI tick-off.
+
+Call this AFTER the planner skill has written a real plan to `/mnt/user-data/workspace/plan.json`. Do NOT call it if the planner returned `{"status":"skip",...}` — that's a trivial request and there's nothing to track.
+
+## Args
+
+- `todos`: a list of `{content: str, status: str}` items, one per plan.json step. The first item's status is `"in_progress"` (you're about to execute it); subsequent items are `"pending"`.
+
+## Lifecycle
+
+1. **Initial hydration** (after plan.json write): one todo per step, first `in_progress`, rest `pending`.
+2. **Mid-execution** (between steps): the previous step flips to `"completed"`, the current step flips to `"in_progress"`. Send the full updated array each time — the tool replaces state, doesn't merge.
+3. **End** (after the last step): the last item flips to `"completed"`.
+
+## Statuses
+
+- `pending`: not yet started.
+- `in_progress`: actively being worked on. Keep exactly one at a time unless the plan marked steps as parallel.
+- `completed`: finished. Only mark this when the step's `produces[]` are actually present.
+"""
+
+
+class ArgusTodoMiddleware(TodoMiddleware):
+    """``TodoMiddleware`` variant whose default prompts defer to the planner skill.
+
+    Behavior (context-loss reminders, premature-exit prevention) inherits
+    unchanged from the parent class. Only the user-facing system prompt and
+    tool description differ.
+
+    Callers can still override either prompt explicitly via the constructor
+    kwargs; the Argus defaults apply only when no value is supplied.
+    """
+
+    def __init__(
+        self,
+        system_prompt: str | None = None,
+        tool_description: str | None = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(
+            system_prompt=system_prompt if system_prompt is not None else _ARGUS_SYSTEM_PROMPT,
+            tool_description=tool_description if tool_description is not None else _ARGUS_TOOL_DESCRIPTION,
+            **kwargs,
+        )

backend/tests/test_argus_todo_middleware.py

@@ -0,0 +1,137 @@
+"""Tests for ArgusTodoMiddleware and the factory routing in lead_agent.
+
+The subclass should:
+  1. Default to Argus-aligned system prompt + tool description.
+  2. Honour explicit kwargs that override the defaults.
+  3. Inherit behavioural hooks from TodoMiddleware (smoke check).
+
+The factory (``_create_todo_list_middleware``) should:
+  4. Return ArgusTodoMiddleware when agent_name == "qwen-local-coder".
+  5. Return a vanilla TodoMiddleware (with the upstream prompt) for any
+     other agent_name.
+  6. Return None when is_plan_mode is False, regardless of agent_name.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from langchain_core.messages import AIMessage, HumanMessage
+
+from deerflow.agents.lead_agent.agent import _create_todo_list_middleware
+from deerflow.agents.middlewares.argus_todo_middleware import (
+    ArgusTodoMiddleware,
+    _ARGUS_SYSTEM_PROMPT,
+    _ARGUS_TOOL_DESCRIPTION,
+)
+from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
+
+
+# ---------------------------------------------------------------------------
+# Construction defaults
+# ---------------------------------------------------------------------------
+
+
+def test_default_construction_uses_argus_prompt():
+    """Without args the subclass picks up the planner-aligned prompt."""
+    mw = ArgusTodoMiddleware()
+    # The base class stores the system prompt; we check it lands there.
+    # Implementation detail: TodoListMiddleware exposes the prompt via
+    # `tools[0].description` and the system prompt via attached state — we
+    # check both layers by sampling text we know is unique to the Argus
+    # variant.
+    assert "planner" in _ARGUS_SYSTEM_PROMPT.lower()
+    assert "/mnt/skills/public/planner/SKILL.md" in _ARGUS_SYSTEM_PROMPT
+    assert "supersedes" in _ARGUS_SYSTEM_PROMPT
+    # The middleware itself must be alive after construction.
+    assert isinstance(mw, ArgusTodoMiddleware)
+    assert isinstance(mw, TodoMiddleware)
+
+
+def test_default_tool_description_mentions_plan_json():
+    assert "plan.json" in _ARGUS_TOOL_DESCRIPTION
+    assert "skip" in _ARGUS_TOOL_DESCRIPTION
+
+
+def test_explicit_overrides_take_priority():
+    """An explicit prompt or description in kwargs overrides the default."""
+    custom_prompt = "<custom_prompt>just for testing</custom_prompt>"
+    custom_tool = "Custom tool description for testing only."
+    mw = ArgusTodoMiddleware(
+        system_prompt=custom_prompt,
+        tool_description=custom_tool,
+    )
+    # We can't easily introspect the system prompt without instantiating an
+    # agent. Confirm the constructor accepted the kwargs without error and
+    # produced the right type.
+    assert isinstance(mw, ArgusTodoMiddleware)
+
+
+# ---------------------------------------------------------------------------
+# Inherited behaviour smoke-check
+# ---------------------------------------------------------------------------
+
+
+def _make_runtime():
+    runtime = MagicMock()
+    runtime.context = {"thread_id": "test-thread"}
+    return runtime
+
+
+def test_inherits_context_loss_reminder_path():
+    """The before_model context-loss reminder logic comes from
+    TodoMiddleware. Smoke-check that it still fires for the subclass when
+    todos exist in state but the original write_todos call has rolled out
+    of context."""
+    mw = ArgusTodoMiddleware()
+    state = {
+        "todos": [
+            {"status": "in_progress", "content": "step 1"},
+            {"status": "pending", "content": "step 2"},
+        ],
+        "messages": [
+            # An old AIMessage with NO write_todos tool call — simulates
+            # the call having scrolled out of the truncated context.
+            AIMessage(content="some prior assistant text", tool_calls=[]),
+        ],
+    }
+    result = mw.before_model(state, _make_runtime())
+    assert result is not None, "expected a reminder when todos exist but write_todos is gone"
+    assert "messages" in result
+    injected = result["messages"][0]
+    assert isinstance(injected, HumanMessage)
+    assert getattr(injected, "name", None) == "todo_reminder"
+
+
+# ---------------------------------------------------------------------------
+# Factory routing
+# ---------------------------------------------------------------------------
+
+
+def test_factory_returns_argus_for_qwen_local_coder():
+    mw = _create_todo_list_middleware(is_plan_mode=True, agent_name="qwen-local-coder")
+    assert isinstance(mw, ArgusTodoMiddleware)
+
+
+def test_factory_returns_vanilla_for_other_agents():
+    """Any agent name other than qwen-local-coder gets the upstream
+    TodoMiddleware. Confirm by type — ArgusTodoMiddleware is a subclass,
+    so we check the exact type rather than isinstance."""
+    mw = _create_todo_list_middleware(is_plan_mode=True, agent_name="code-reviewer")
+    assert isinstance(mw, TodoMiddleware)
+    assert not isinstance(mw, ArgusTodoMiddleware), (
+        "code-reviewer should keep the upstream prompt, not Argus's override"
+    )
+
+
+def test_factory_returns_vanilla_when_agent_name_is_empty():
+    """No agent_name (the default) means no override — straight upstream."""
+    mw = _create_todo_list_middleware(is_plan_mode=True)
+    assert isinstance(mw, TodoMiddleware)
+    assert not isinstance(mw, ArgusTodoMiddleware)
+
+
+def test_factory_returns_none_when_plan_mode_off():
+    """Even for qwen-local-coder, plan_mode off means no middleware."""
+    mw = _create_todo_list_middleware(is_plan_mode=False, agent_name="qwen-local-coder")
+    assert mw is None