feat: introduce agent mode functionality and enhance backend handling

- Added support for `agent_mode` in `ChatRequest` and `build_agent`, allowing for "chat" and "research" modes to control agent behavior and subagent inclusion.
- Implemented `_apply_mode_harness_profile` to register provider harness profiles based on the selected agent mode.
- Refactored `build_system_prompt` to accommodate workspace rules without exposing host paths.
- Enhanced event handling in `should_emit_langgraph_event` to differentiate behavior based on agent mode.
- Updated tests to validate new agent mode functionality and ensure proper integration across components.
- Improved UI components to reflect changes in agent mode settings and interactions.

Author: knqiufan

agent/app/agent.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+import threading
+from typing import Any, Literal
 
 from app.config import get_settings
 from app.prompts import (
@@ -13,27 +14,38 @@
     SYSTEM_PROMPT,
     build_system_prompt,
 )
+from app.workspace_backend import make_workspace_backend_factory
 
 logger = logging.getLogger(__name__)
 
+AgentMode = Literal["chat", "research"]
+
+_HARNESS_LOCK = threading.Lock()
+_PROVIDER_PROFILE_KEYS = (
+    "openai",
+    "anthropic",
+    "google",
+    "google_genai",
+    "azure_openai",
+    "bedrock",
+)
+
 
 def build_agent(
     session_id: str | None = None,
     working_dir: str | None = None,
     checkpointer=None,
     store=None,
     tools: list[Any] | None = None,
-    include_subagents: bool = True,
+    include_subagents: bool | None = None,
+    agent_mode: AgentMode = "chat",
     model: Any | None = None,
 ):
     """Create a DeepAgent graph for the given session context.
 
-    ``model`` may be a LangChain chat model instance or a provider:model string.
-    When omitted, falls back to ``settings.agent_model``.
-
-    When ``working_dir`` validates, filesystem tools (`ls` / `read_file` / …)
-    and shell `execute` use that directory as the default root — not an
-    ephemeral in-memory StateBackend.
+    ``agent_mode="chat"`` (default) disables the ``task`` tool and all sync
+    subagents, including DeepAgents' auto-injected general-purpose agent.
+    ``agent_mode="research"`` enables researcher/coder/analyst plus GP.
     """
     del session_id  # reserved for future per-session customization
     settings = get_settings()
@@ -50,16 +62,15 @@ def build_agent(
             "deepagents is not installed. Install with: pip install -e '.[agent]'"
         ) from exc
 
+    mode = _resolve_agent_mode(agent_mode, include_subagents)
     validated = _resolve_working_dir(working_dir)
     agent_tools = tools if tools is not None else _safe_get_tools()
-    subagents = _build_subagents(settings) if include_subagents else []
+    subagents = _build_subagents(settings) if mode == "research" else []
 
     kwargs: dict[str, Any] = {
         "model": model if model is not None else settings.agent_model,
         "tools": agent_tools,
-        "system_prompt": build_system_prompt(
-            working_dir=str(validated) if validated is not None else None
-        ),
+        "system_prompt": build_system_prompt(has_workspace=validated is not None),
         "subagents": subagents,
         "interrupt_on": settings.interrupt_config,
         "checkpointer": checkpointer,
@@ -68,25 +79,43 @@ def build_agent(
 
     kwargs["skills"] = [str(settings.skills_dir)]
     kwargs["memory"] = [str(settings.memories_dir)]
-    kwargs["backend"] = _make_backend_factory(validated, LocalShellBackend, StateBackend, CompositeBackend)
-    return create_deep_agent(**kwargs)
+    kwargs["backend"] = make_workspace_backend_factory(
+        validated, LocalShellBackend, StateBackend, CompositeBackend
+    )
 
+    with _HARNESS_LOCK:
+        _apply_mode_harness_profile(mode)
+        return create_deep_agent(**kwargs)
 
-def _make_backend_factory(validated, local_shell_cls, state_cls, composite_cls):
-    """Bind filesystem + shell to working_dir when present; else StateBackend."""
 
-    def factory(rt):
-        if validated is not None:
-            # virtual_mode: `/` and relative paths map under working_dir.
-            # LocalShellBackend also sets shell cwd to working_dir.
-            return local_shell_cls(
-                root_dir=str(validated),
-                virtual_mode=True,
-                inherit_env=True,
-            )
-        return composite_cls(default=state_cls(rt), routes={})
+def _resolve_agent_mode(
+    agent_mode: AgentMode,
+    include_subagents: bool | None,
+) -> AgentMode:
+    if include_subagents is None:
+        return "research" if agent_mode == "research" else "chat"
+    # Backward-compatible override used by older tests/callers.
+    return "research" if include_subagents else "chat"
+
+
+def _apply_mode_harness_profile(mode: AgentMode) -> None:
+    """Register provider harness profiles so chat mode truly drops ``task``."""
+    try:
+        from deepagents import (
+            GeneralPurposeSubagentProfile,
+            HarnessProfileConfig,
+            register_harness_profile,
+        )
+    except ImportError:
+        logger.warning("Harness profile APIs unavailable; cannot disable general-purpose")
+        return
 
-    return factory
+    enabled = mode == "research"
+    profile = HarnessProfileConfig(
+        general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=enabled)
+    )
+    for key in _PROVIDER_PROFILE_KEYS:
+        register_harness_profile(key, profile)
 
 
 def _resolve_working_dir(working_dir: str | None):
@@ -152,4 +181,10 @@ def _build_subagents(settings) -> list[dict[str, Any]]:
 
 
 # Re-export for tests that imported SYSTEM_PROMPT via this module historically.
-__all__ = ["SYSTEM_PROMPT", "build_agent", "_build_subagents", "_resolve_working_dir"]
+__all__ = [
+    "SYSTEM_PROMPT",
+    "build_agent",
+    "_build_subagents",
+    "_resolve_working_dir",
+    "_apply_mode_harness_profile",
+]

agent/app/models.py

@@ -80,6 +80,8 @@ class ChatRequest(BaseModel):
     config: ChatConfig = Field(default_factory=ChatConfig)
     session_id: str | None = None
     working_dir: str | None = None
+    # chat = direct tools only; research = DeepAgents task/subagents
+    agent_mode: str = "chat"
 
     model_config = {"frozen": False, "extra": "ignore"}
 

agent/app/prompts.py

@@ -20,22 +20,27 @@
 """
 
 
-def build_system_prompt(working_dir: str | None = None) -> str:
-    """Assemble the system prompt, including working-directory path rules."""
-    if not working_dir:
+def build_system_prompt(has_workspace: bool = False, working_dir: str | None = None) -> str:
+    """Assemble the system prompt, including workspace virtual-path rules.
+
+    ``working_dir`` is accepted for backward compatibility but is never injected
+    into the prompt (host absolute paths cause models to concatenate badly).
+    """
+    del working_dir
+    if not has_workspace:
         return SYSTEM_PROMPT
 
-    workspace_rules = f"""
+    workspace_rules = """
 
 ## Working directory
-- The bound project directory is: `{working_dir}`
+- The bound project is mounted at the virtual root `/workspace`.
 - Filesystem tools (`ls`, `read_file`, `write_file`, `edit_file`, `glob`, `grep`)
-  treat `/` as this directory. Prefer virtual paths like `/`, `/src`, `/README.md`.
-- Shell `execute` also starts with this directory as its cwd.
-- Do NOT invent a `/workspace` prefix and do NOT concatenate `/workspace` with
-  Windows absolute paths (e.g. `/workspaceD:\\...` is wrong).
-- Prefer relative or virtual paths over host absolute paths unless the user
-  explicitly asks for a path outside the project.
+  MUST use paths like `/workspace`, `/workspace/src`, `/workspace/README.md`.
+- Shell `execute` already starts with the project directory as its cwd; prefer
+  relative shell paths (e.g. `ls`, `type README.md`) rather than host paths.
+- NEVER use Windows/host absolute paths (e.g. `D:\\...`).
+- NEVER concatenate `/workspace` with a host path
+  (e.g. `/workspaceD:\\...` is invalid).
 """
     return SYSTEM_PROMPT + workspace_rules
 
@@ -46,6 +51,7 @@ def build_system_prompt(working_dir: str | None = None) -> str:
 - Gather and synthesize information thoroughly.
 - Prefer verifiable sources and explicit uncertainty when evidence is weak.
 - Use memory search when prior findings may help.
+- When filesystem tools are available, use only `/workspace/<relative-path>`.
 
 ## Output
 - Summarize findings clearly.
@@ -59,6 +65,7 @@ def build_system_prompt(working_dir: str | None = None) -> str:
 - Write clean, maintainable code with clear error handling.
 - Prefer small, reviewable changes over large rewrites.
 - Use workspace filesystem tools carefully when available.
+- Filesystem paths must be `/workspace/<relative-path>` only.
 
 ## Output
 - Provide complete, runnable snippets when asked.

agent/app/routers/agent.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import traceback
 from collections.abc import AsyncGenerator
 from typing import Any
 
@@ -39,7 +40,11 @@ async def agent_chat(request: ChatRequest) -> ChatResponse:
     except HTTPException:
         raise
     except Exception as exc:
-        logger.exception("agent chat failed")
+        logger.exception(
+            "agent chat failed session_id=%s mode=%s",
+            request.session_id,
+            request.agent_mode,
+        )
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
 
@@ -58,43 +63,57 @@ async def agent_stream(request: ChatRequest):
 
 
 async def _stream_agent(request: ChatRequest) -> AsyncGenerator[str, None]:
+    open_tools: dict[str, dict[str, Any]] = {}
+    thinking_acc = ""
+    emitted_tool_ids: set[str] = set()
     try:
         agent = _build_request_agent(request)
-        thinking_acc = ""
-        emitted_tool_ids: set[str] = set()
         async for event in agent.astream_events(
             {"messages": _request_to_agent_messages(request)},
             config=_thread_config(request),
             version="v2",
         ):
-            if not should_emit_langgraph_event(event):
+            if not should_emit_langgraph_event(event, agent_mode=request.agent_mode):
                 continue
             for sse_data in _format_sse_events(
                 event,
                 thinking_acc=thinking_acc,
                 emitted_tool_ids=emitted_tool_ids,
+                open_tools=open_tools,
             ):
-                # Keep cumulative thinking for suffix-delta extraction.
                 if sse_data.startswith("event: thinking\n"):
                     data_line = sse_data.split("\n", 2)[1]
                     payload = json.loads(data_line.removeprefix("data: "))
                     thinking_acc += str(payload.get("content") or "")
                 yield sse_data
+        for frame in _close_open_tools(open_tools, reason="Stream ended without tool_end"):
+            yield frame
         yield _sse("done", {"finished": True})
     except Exception as exc:
-        logger.exception("agent stream failed")
+        tb = traceback.format_exc()
+        logger.error(
+            "agent stream failed session_id=%s mode=%s error=%s\n%s",
+            request.session_id,
+            request.agent_mode,
+            exc,
+            tb,
+        )
+        for frame in _close_open_tools(open_tools, reason=str(exc) or "stream error"):
+            yield frame
         yield _sse("error", {"message": str(exc)})
 
 
 def _build_request_agent(request: ChatRequest):
     """Assemble agent using the provider binding from the chat request."""
     model = resolve_chat_model(request.config)
+    mode = "research" if request.agent_mode == "research" else "chat"
     return build_agent(
         session_id=request.session_id,
         working_dir=request.working_dir,
         checkpointer=get_checkpointer(),
         store=get_store(),
         model=model,
+        agent_mode=mode,
     )
 
 
@@ -104,9 +123,11 @@ def _request_to_agent_messages(request: ChatRequest) -> list[dict[str, str]]:
 
 def _thread_config(request: ChatRequest) -> dict[str, Any]:
     settings = get_settings()
+    # Keep a finite recursion ceiling for research; chat stays leaner.
+    limit = settings.agent_recursion_limit if request.agent_mode == "research" else 50
     config: dict[str, Any] = {
         "configurable": {"thread_id": request.session_id or "default"},
-        "recursion_limit": settings.agent_recursion_limit,
+        "recursion_limit": limit,
     }
     if request.config.model:
         config["configurable"]["model"] = request.config.model
@@ -118,11 +139,13 @@ def _format_sse_events(
     *,
     thinking_acc: str = "",
     emitted_tool_ids: set[str] | None = None,
+    open_tools: dict[str, dict[str, Any]] | None = None,
 ) -> list[str]:
     """Convert a LangGraph event into zero or more SSE payloads."""
     kind = event.get("event")
     run_id = event.get("run_id") or event.get("id")
     tool_ids = emitted_tool_ids if emitted_tool_ids is not None else set()
+    open_map = open_tools if open_tools is not None else {}
 
     if kind == "on_chat_model_stream":
         chunk = event.get("data", {}).get("chunk")
@@ -146,18 +169,47 @@ def _format_sse_events(
             return []
         if tool_id:
             tool_ids.add(tool_id)
+            open_map[tool_id] = {
+                "name": payload.get("name") or "",
+                "server_id": payload.get("server_id"),
+            }
         return [_sse("tool_start", payload)]
 
     if kind == "on_tool_end":
+        tool_id = str(run_id) if run_id else ""
+        open_map.pop(tool_id, None)
         return [_format_tool_end(event, run_id, error=None)]
 
     if kind == "on_tool_error":
+        tool_id = str(run_id) if run_id else ""
+        open_map.pop(tool_id, None)
         err = event.get("data", {}).get("error") or event.get("data", {}).get("message")
         return [_format_tool_end(event, run_id, error=str(err or "tool error"))]
 
     return []
 
 
+def _close_open_tools(
+    open_tools: dict[str, dict[str, Any]],
+    *,
+    reason: str,
+) -> list[str]:
+    frames: list[str] = []
+    for tool_id, meta in list(open_tools.items()):
+        payload: dict[str, Any] = {
+            "id": tool_id,
+            "name": str(meta.get("name") or "unknown"),
+            "output": "",
+            "status": "error",
+            "error": reason,
+        }
+        if meta.get("server_id"):
+            payload["server_id"] = meta["server_id"]
+        frames.append(_sse("tool_end", payload))
+        open_tools.pop(tool_id, None)
+    return frames
+
+
 def _format_tool_end(event: dict[str, Any], run_id: Any, error: str | None) -> str:
     name = str(event.get("name") or "")
     output = event.get("data", {}).get("output", "")
@@ -168,7 +220,6 @@ def _format_tool_end(event: dict[str, Any], run_id: Any, error: str | None) -> s
         "status": "error" if error else "complete",
     }
     if name == "mcp_bridge":
-        # Prefer the expanded tool name from inputs when available.
         raw_input = event.get("data", {}).get("input")
         if isinstance(raw_input, dict):
             payload["name"] = str(