fix(mcp): bring-up has no timeout and externalized tool outputs are counted as undelivered artifacts (#4657)

* fix: bound MCP server bring-up timeouts and exclude externalized tool outputs from delivery verification

Two related robustness fixes:

1. MCP server bring-up was unbounded. tool_call_timeout only covered
   session.call_tool(); tool discovery (subprocess spawn + initialize +
   tools/list) and persistent stdio session initialization could hang
   forever, blocking agent construction (and on the Gateway event loop,
   the whole process). Add a per-server session_init_timeout
   (default DEFAULT_MCP_SESSION_INIT_TIMEOUT = 60s, null disables) that
   bounds both discovery and pooled-session initialization. The session
   pool's existing cancellation handling tears down a session stuck
   mid-creation in its own task.

2. ToolOutputBudgetMiddleware externalizes oversized tool outputs into
   outputs/.tool-results/ (configurable tool_output.storage_subdir). The
   workspace-change scanner and run delivery verification counted those
   files as produced artifacts, so any run that externalized a tool output
   without also presenting a real artifact failed with
   "Artifact delivery incomplete". Exclude TOOL_RESULTS_DIRNAME via a
   shared constant (mirroring BROWSER_FRAMES_DIRNAME) and thread the
   configured storage_subdir through snapshot capture so both
   workspace-changes events and delivery verification stay clean.

* review: enforce single-segment tool_output.storage_subdir; document discovery-timeout cleanup

Address review feedback:

1. A custom tool_output.storage_subdir with a path separator (e.g.
   cache/tool-results) silently no-oped the workspace-scanner exclusion:
   os.walk yields one-segment dirnames, so a nested value never matched and
   its files were counted as produced artifacts again. ToolOutputConfig now
   validates storage_subdir as a single directory name (rejects separators,
   .., absolute, empty) with tests, so the exclusion is always sound.

2. The discovery-timeout path now documents why cancellation is safe, mirroring
   the session-init note: discovery runs inside the adapter's nested async
   context managers, and stdio_client's finally terminates the process tree
   (SIGTERM->SIGKILL on POSIX, process-tree on Windows), so a timed-out npx
   subprocess and its children are reaped rather than accumulating.

* review: log session-init timeouts and align API response model default with runtime config

Address second-round review feedback:

1. A session-init timeout raised TimeoutError without any log, unlike the
   discovery timeout which logs a WARNING. Wrap the bounded get_session in a
   try/except that logs the timeout (server name + seconds) and re-raises, so
   operators can diagnose tool-call failures caused by hung MCP sessions.

2. McpServerConfigResponse.session_init_timeout defaulted to None while
   McpServerConfig defaults to 60s: a server created via PUT /api/mcp/config
   without the field was persisted with null (no timeout) while the same
   server created in the config file got 60s. Align the response-model default
   to DEFAULT_MCP_SESSION_INIT_TIMEOUT so API-created and file-created servers
   behave the same; an explicit null still opts out.

* review: narrow the discovery-timeout handler to the bounded wait_for path

The except TimeoutError clause covered both the bounded wait_for branch and
the bare discovery branch. With session_init_timeout opted out (None), a
TimeoutError raised by discovery itself would hit the %.1f format with None:
logging raises TypeError internally, the WARNING is silently dropped, and a
--- Logging error --- traceback goes to stderr.

Narrow the handler to wrap only the wait_for call, where the branch condition
guarantees the timeout value is not None. A discovery-internal TimeoutError on
the opted-out path now falls through to the generic failure handler and is
reported as 'tool discovery failed' with exc_info. Covered by a regression
test that asserts the skip is reported without any broken format.

CHANGELOG.md

@@ -533,6 +533,17 @@ This section accumulates work toward the **2.1.0** milestone
   thread. ([#4394])
 - **tools:** Exclude injected runtime from the `list_uploaded_files` schema.
   ([#4376])
+- **mcp:** Bound MCP server bring-up — tool discovery (subprocess spawn +
+  `initialize` + `tools/list`) and persistent stdio session initialization —
+  with a new per-server `session_init_timeout` (default 60s, `null` disables),
+  so a hung stdio server can no longer block agent construction, or the whole
+  Gateway event loop, indefinitely. `tool_call_timeout` still bounds individual
+  stdio tool calls.
+- **runtime:** Tool-output budget externalization no longer trips run delivery
+  verification. The default `.tool-results` storage dir (and any custom
+  `tool_output.storage_subdir`) is excluded from workspace-change snapshots and
+  produced-artifact detection, so a run that only externalized oversized tool
+  outputs succeeds instead of failing as an error.
 
 ### Performance
 

backend/AGENTS.md

@@ -393,7 +393,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 **Shared runtime base** (`build_lead_runtime_middlewares`; subagents reuse most of this via `build_subagent_runtime_middlewares`):
 
 1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages. `additional_kwargs.original_user_content` is server-owned provenance: Gateway strips caller-supplied values for non-internal run requests, trusted IM calls may carry the string they captured before adding transport/file context, and the middleware replaces any non-string value before wrapping. Uploads and sanitization retain first-writer-wins only for validated strings.
-2. **ToolOutputBudgetMiddleware** - Caps tool output size (per app config) before it re-enters the model context
+2. **ToolOutputBudgetMiddleware** - Caps tool output size (per app config) before it re-enters the model context. Oversized results are externalized to `tool_output.storage_subdir` (default `.tool-results`, shared constant `TOOL_RESULTS_DIRNAME`) under the thread outputs dir with a typed synopsis + `read_file` reference left in context; those files are process feedback, so the workspace-changes scanner excludes that directory and run delivery verification never counts them as produced artifacts
 3. **ToolResultSanitizationMiddleware** - Neutralizes framework/injection tags (e.g. `<system-reminder>`) and boundary markers in *remote-content* tool results (`web_fetch`/`web_search`/`image_search`/`web_capture`) so attacker-controlled fetched pages cannot forge trusted framework context. Mirrors `InputSanitizationMiddleware`'s user-input guardrail for the other untrusted-content entry point; sits inner of `ToolOutputBudgetMiddleware` (neutralizes the raw output, then the budget truncates). Local tool output (bash/read_file) is left untouched. Scope is a name-based allowlist, so MCP remote-content tools registered under other names (e.g. `fetch_url`) are not yet covered — a metadata-tagging follow-up is tracked in the middleware source
 4. **ThreadDataMiddleware** - Creates per-thread directories under the user's isolation scope (`backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/{workspace,uploads,outputs}`); resolves identity via `resolve_runtime_user_id(runtime)`, including Gateway runtime context and standalone LangGraph Server auth, then falls back to the request ContextVar / `"default"`
 5. **UploadsMiddleware** - Tracks and injects newly uploaded files into conversation (lead agent only); upload existence checks use the same runtime-resolved user bucket as thread-data creation
@@ -587,7 +587,13 @@ captures a pre-run and post-run snapshot of the thread-owned `workspace` and
 `asyncio.to_thread` and writes a `workspace_changes` event with category
 `workspace` when changes exist. Uploads are intentionally excluded. Text diffs
 are size-limited; binary, large, and sensitive-looking paths are persisted as
-metadata only.
+metadata only. Internal process-feedback directories never count as changes:
+the scanner's `EXCLUDED_DIR_NAMES` drops `BROWSER_FRAMES_DIRNAME` (transient
+browser screenshots) and `TOOL_RESULTS_DIRNAME` (the tool-output budget
+middleware's default externalization subdir, `constants.py` is the shared
+source of truth for both writers and the scanner), and the worker threads the
+configured `tool_output.storage_subdir` through the snapshot capture as an
+extra excluded dir name so custom storage locations stay excluded too.
 
 **Run delivery receipts**: `RunJournal` records each non-empty artifact update
 once per tool `Command` for the terminal `run.delivery` event. When a command
@@ -607,9 +613,14 @@ terminal run status. A receipt failure is retried on a short bounded schedule
 while the owning worker still knows the real outcome and holds the lease. The
 worker derives delivery requirements from the run's workspace snapshots rather
 than a client request option: every regular file created or modified under
-`/mnt/user-data/outputs` is a candidate produced artifact. At least one candidate
-must be covered by a path attributed by the journal to `present_files`;
-presenting only an unrelated pre-existing path does not satisfy delivery.
+`/mnt/user-data/outputs` is a candidate produced artifact. Internal
+process-feedback files are not candidates: the snapshot capture excludes the
+scanner's `EXCLUDED_DIR_NAMES` (including the default tool-output
+externalization subdir) plus the configured `tool_output.storage_subdir`, so a
+run that only externalized oversized tool outputs does not fail delivery. At
+least one candidate must be covered by a path attributed by the journal to
+`present_files`; presenting only an unrelated pre-existing path does not
+satisfy delivery.
 Receipts for such runs add `produced_paths`, `presented_paths`, `matched_paths`,
 `verification`, `stage`, and `satisfied` to the Slice 1 fact fields. Missing a
 matching presentation becomes a run error; a successful presentation is also
@@ -1315,7 +1326,7 @@ Config is env-driven like the others — `MonocleTracingConfig`, built in `get_t
 - `memory` - Memory system (enabled, storage_path, debounce_seconds, shutdown_flush_timeout_seconds, model_name, max_facts, fact_confidence_threshold, injection_enabled, max_injection_tokens, staleness_review_enabled, staleness_age_days, staleness_min_candidates, staleness_max_removals_per_cycle, staleness_protected_categories, staleness_max_lifetime_multiplier, staleness_max_extension_days)
 
 **`extensions_config.json`**:
-- `mcpServers` - Map of server name → config (enabled, type, command, args, env, url, headers, oauth, description, `routing`, `tools`, `tool_call_timeout`). `routing.mode="prefer"` emits `<mcp_routing_hints>` prompt guidance; if `tool_search` defers the hinted tool, `McpRoutingMiddleware` can also auto-promote matching deferred schemas before the model call. It does not hard-disable other tools.
+- `mcpServers` - Map of server name → config (enabled, type, command, args, env, url, headers, oauth, description, `routing`, `tools`, `tool_call_timeout`, `session_init_timeout`). `routing.mode="prefer"` emits `<mcp_routing_hints>` prompt guidance; if `tool_search` defers the hinted tool, `McpRoutingMiddleware` can also auto-promote matching deferred schemas before the model call. It does not hard-disable other tools. `session_init_timeout` (default `DEFAULT_MCP_SESSION_INIT_TIMEOUT` = 60s, `null` to disable) bounds server bring-up: tool discovery and persistent stdio session initialization, so a hung server cannot block agent construction indefinitely. `tool_call_timeout` bounds individual stdio tool calls.
 - `tool_search.auto_promote_top_k` - Global MCP routing auto-promote breadth. Default `3`, clamped to `1..5`; applies only when `tool_search.enabled=true` and only to deferred MCP tools with `routing.mode="prefer"` and non-empty keywords. For lead agents the deferred catalog is built from the full configured MCP set; auto-promotion never grants authority because an active skill's runtime policy still filters model-visible schemas, `tool_search` results, and execution.
 - `skills` - Map of skill name → state (enabled)
 - `middlewares` - Zero-argument `AgentMiddleware` class paths for lead and subagent runtime extension. `config.yaml -> extensions` can override these fields after validation; overrides are replace-per-field, not list concatenation.

backend/app/gateway/routers/mcp.py

@@ -20,6 +20,7 @@ from deerflow.config.extensions_config import (
     normalize_mcp_transport_alias,
     reload_extensions_config,
 )
+from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT
 from deerflow.mcp.cache import reset_mcp_tools_cache
 
 logger = logging.getLogger(__name__)
@@ -379,6 +380,10 @@ class McpServerConfigResponse(BaseModel):
     tools: dict[str, McpToolOverride] = Field(default_factory=dict, description="Per-original-tool MCP configuration overrides")
     tool_name_prefix: bool = Field(default=True, description="Whether to prefix discovered tool names with the MCP server name")
     tool_call_timeout: float | None = Field(default=None, description="Timeout in seconds for individual stdio MCP tool calls")
+    # Default matches McpServerConfig: this model's defaults feed model_dump()
+    # into the persisted extensions config on PUT, so an API-created server that
+    # omits the field must get the same bring-up timeout as a file-created one.
+    session_init_timeout: float | None = Field(default=DEFAULT_MCP_SESSION_INIT_TIMEOUT, description="Timeout in seconds for MCP server bring-up (tool discovery and persistent stdio session initialization); null means no timeout")
     model_config = ConfigDict(extra="allow")
 
     @model_validator(mode="before")

backend/docs/MCP_SERVER.md

@@ -96,9 +96,14 @@ backward compatibility. Disable it only when every resulting tool name remains
 unique across the enabled servers. Stdio tools continue to use DeerFlow's
 persistent per-thread session pool regardless of this setting.
 
-## Per-Tool Timeout (Stdio MCP Servers)
+## Server Timeouts (Stdio MCP Servers)
 
-For `stdio` MCP servers, set `tool_call_timeout` to limit each individual MCP tool call in seconds:
+Two independent timeouts bound stdio MCP servers. `session_init_timeout` covers
+server bring-up — tool discovery (subprocess spawn + `initialize` +
+`tools/list`) and persistent-session initialization — and defaults to 60s so a
+hung server (e.g. `npx` blocked on a package download, or a server that never
+answers `initialize`) cannot block agent construction indefinitely. Set it to
+`null` to disable:
 
 ```json
 {
@@ -111,13 +116,17 @@ For `stdio` MCP servers, set `tool_call_timeout` to limit each individual MCP to
          "env": {
             "GITHUB_TOKEN": "$GITHUB_TOKEN"
          },
+         "session_init_timeout": 60,
          "tool_call_timeout": 60
       }
    }
 }
 ```
 
-`tool_call_timeout` only applies to `stdio` servers. `http` and `sse` servers use transport-level timeouts, and DeerFlow logs a warning if `tool_call_timeout` is configured for those transports.
+`tool_call_timeout` limits each individual tool call in seconds and applies only
+to `stdio` servers; `http` and `sse` servers use transport-level timeouts, and
+DeerFlow logs a warning if `tool_call_timeout` is configured for those
+transports.
 
 ## Filesystem MCP Servers
 

backend/packages/harness/deerflow/config/extensions_config.py

@@ -12,6 +12,7 @@ from typing import Any, Literal
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from deerflow.config.runtime_paths import existing_project_file
+from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT
 
 logger = logging.getLogger(__name__)
 
@@ -106,6 +107,14 @@ class McpServerConfig(BaseModel):
         default=None,
         description="Timeout in seconds for individual stdio MCP tool calls. HTTP/SSE servers use transport-level timeouts. None means no timeout.",
     )
+    session_init_timeout: float | None = Field(
+        default=DEFAULT_MCP_SESSION_INIT_TIMEOUT,
+        description=(
+            "Timeout in seconds for MCP server bring-up: tool discovery (subprocess spawn + initialize + tools/list) "
+            "and persistent stdio session initialization. Defaults to DEFAULT_MCP_SESSION_INIT_TIMEOUT so a hung "
+            "server cannot block agent construction indefinitely. None means no timeout."
+        ),
+    )
     model_config = ConfigDict(extra="allow")
 
     @model_validator(mode="before")

backend/packages/harness/deerflow/config/tool_output_config.py

@@ -2,7 +2,11 @@
 
 from __future__ import annotations
 
-from pydantic import BaseModel, Field
+import os
+
+from pydantic import BaseModel, Field, field_validator
+
+from deerflow.constants import TOOL_RESULTS_DIRNAME
 
 
 class ToolOutputConfig(BaseModel):
@@ -49,9 +53,31 @@ class ToolOutputConfig(BaseModel):
         description="Tail characters for fallback truncation.",
     )
     storage_subdir: str = Field(
-        default=".tool-results",
-        description="Subdirectory under the thread outputs path for persisted tool results.",
+        default=TOOL_RESULTS_DIRNAME,
+        description=(
+            "Single-segment directory name under the thread outputs path for persisted tool results. "
+            "TOOL_RESULTS_DIRNAME is always excluded by the workspace-changes scanner; other custom values are "
+            "excluded from workspace snapshots and run delivery verification at capture time."
+        ),
     )
+
+    @field_validator("storage_subdir")
+    @classmethod
+    def _storage_subdir_is_single_segment(cls, value: str) -> str:
+        """Require a single directory name (no path separators).
+
+        The workspace-changes scanner prunes by directory name during
+        ``os.walk``, which yields one-segment dirnames — a nested value like
+        ``cache/tool-results`` would never match the exclusion and its files
+        would silently be counted as produced artifacts again. A loud config
+        error beats a silent exclusion no-op.
+        """
+        if value == "" or value in {".", ".."} or os.path.isabs(value):
+            raise ValueError("storage_subdir must be a single non-empty directory name")
+        if "/" in value or "\\" in value:
+            raise ValueError(f"storage_subdir must be a single directory name without path separators (got {value!r})")
+        return value
+
     exempt_tools: list[str] = Field(
         default_factory=lambda: ["read_file", "read_file_tool"],
         description="Tool names exempt from budget enforcement (prevents persist→read→persist loops).",

backend/packages/harness/deerflow/constants.py

@@ -9,6 +9,25 @@ DEFAULT_SKILLS_CONTAINER_PATH = "/mnt/skills"
 # this single source of truth so the name cannot drift between them.
 BROWSER_FRAMES_DIRNAME = ".browser-frames"
 
+# Default subdirectory (under a thread's outputs dir) where the tool-output
+# budget middleware persists oversized tool outputs. These are process
+# feedback the model reads back via ``read_file`` (the budget preview carries
+# the reference), not deliverables, so the workspace-changes scanner excludes
+# this directory and run delivery verification never counts it as a produced
+# artifact. Both the budget middleware's default ``storage_subdir`` and the
+# scanner import this single source of truth so the name cannot drift between
+# them; a custom configured ``storage_subdir`` is threaded through the
+# snapshot capture as an extra excluded dir name.
+TOOL_RESULTS_DIRNAME = ".tool-results"
+
+# Default timeout (seconds) for MCP server bring-up: tool discovery (subprocess
+# spawn + initialize + tools/list) and persistent-session initialization. A hung
+# stdio server (e.g. npx blocked on a package download or a server that never
+# answers initialize) would otherwise block agent construction forever — and on
+# the Gateway event loop, the whole process. Per-server override is
+# ``mcpServers.<name>.session_init_timeout``; ``None`` disables the timeout.
+DEFAULT_MCP_SESSION_INIT_TIMEOUT = 60.0
+
 # Persisted run-event envelope limits. Runtime definitions and the ORM both
 # import these from this dependency-free module so lower layers never need to
 # initialize deerflow.runtime just to validate storage constraints.

backend/packages/harness/deerflow/mcp/tools.py

@@ -16,6 +16,7 @@ from langgraph.config import get_config
 
 from deerflow.config.extensions_config import ExtensionsConfig, resolve_effective_mcp_routing
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, Paths, get_paths
+from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT
 from deerflow.mcp.client import build_servers_config
 from deerflow.mcp.oauth import build_oauth_tool_interceptor, get_initial_oauth_headers
 from deerflow.mcp.session_pool import get_session_pool
@@ -425,12 +426,30 @@ def _convert_call_tool_result(
     return lc_content, artifact
 
 
+def _resolve_session_init_timeout(server_cfg: Any) -> float | None:
+    """Return the effective session-init timeout for *server_cfg*.
+
+    ``None`` (an explicit opt-out) stays ``None``. Any other non-numeric value
+    falls back to the default rather than being passed to ``asyncio.wait_for``
+    (which would raise on it) or silently disabling the bound: pydantic
+    guarantees a float for real configs, but configs built with mocks in tests
+    can supply anything, and the fallback keeps the hang-protection in place.
+    """
+    value = server_cfg.session_init_timeout if server_cfg is not None else DEFAULT_MCP_SESSION_INIT_TIMEOUT
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return DEFAULT_MCP_SESSION_INIT_TIMEOUT
+    return float(value)
+
+
 def _make_session_pool_tool(
     tool: BaseTool,
     server_name: str,
     connection: dict[str, Any],
     tool_interceptors: list[Any] | None = None,
     tool_call_timeout: float | None = None,
+    session_init_timeout: float | None = None,
     tool_name_prefix: bool = True,
 ) -> BaseTool:
     """Wrap an MCP tool so it reuses a persistent session from the pool.
@@ -495,7 +514,29 @@ def _make_session_pool_tool(
             session_env.setdefault("TMP", str(tmp_dir))
             session_env.setdefault("TEMP", str(tmp_dir))
             session_connection["env"] = session_env
-        session = await pool.get_session(server_name, scope_key, session_connection)
+        if session_init_timeout is not None:
+            # Cancellation here is safe: MCPSessionPool.get_session owns the
+            # teardown of a session stuck mid-creation (it signals close and
+            # waits for the owner task's __aexit__ to run in its own task),
+            # so a hung server cannot leak a session or block the turn.
+            try:
+                session = await asyncio.wait_for(
+                    pool.get_session(server_name, scope_key, session_connection),
+                    timeout=session_init_timeout,
+                )
+            except TimeoutError:
+                # Surface the timeout at the same log level as discovery
+                # timeouts: the tool call still fails with a TimeoutError the
+                # model can react to, but operators need the WARNING to
+                # diagnose tool-call failures caused by hung MCP sessions.
+                logger.warning(
+                    "MCP session initialization for server '%s' timed out after %.1fs",
+                    server_name,
+                    session_init_timeout,
+                )
+                raise
+        else:
+            session = await pool.get_session(server_name, scope_key, session_connection)
 
         # Build common call_tool kwargs once — only add keys when needed so
         # existing call-sites that assert on exact arguments are not affected.
@@ -653,16 +694,50 @@ async def get_mcp_tools() -> list[BaseTool]:
             try:
                 server_cfg = extensions_config.mcp_servers.get(server_name)
                 tool_name_prefix = server_cfg.tool_name_prefix if server_cfg is not None else True
+                session_init_timeout = _resolve_session_init_timeout(server_cfg)
                 if tool_name_prefix:
-                    return await client.get_tools(server_name=server_name)
-                return await load_mcp_tools(
-                    None,
-                    connection=servers_config[server_name],
-                    callbacks=client.callbacks,
-                    server_name=server_name,
-                    tool_interceptors=client.tool_interceptors,
-                    tool_name_prefix=False,
-                )
+                    discovery = client.get_tools(server_name=server_name)
+                else:
+                    discovery = load_mcp_tools(
+                        None,
+                        connection=servers_config[server_name],
+                        callbacks=client.callbacks,
+                        server_name=server_name,
+                        tool_interceptors=client.tool_interceptors,
+                        tool_name_prefix=False,
+                    )
+                if session_init_timeout is not None:
+                    # Timeout tool discovery (subprocess spawn + initialize +
+                    # tools/list) so a hung stdio server cannot block agent
+                    # construction indefinitely. Per-server because the gather
+                    # below runs each server independently — one slow server
+                    # must not prevent the others from contributing tools.
+                    #
+                    # Cancellation here is safe: discovery runs inside the
+                    # adapter's nested async context managers (load_mcp_tools →
+                    # create_session → _create_stdio_session → stdio_client),
+                    # and wait_for's CancelledError unwinds them. stdio_client's
+                    # finally closes stdin, waits for a graceful exit, then
+                    # escalates to _terminate_process_tree (SIGTERM→SIGKILL on
+                    # POSIX, process-tree termination on Windows), so the npx
+                    # subprocess and any children it spawned are reaped — no
+                    # orphan processes accumulate across repeated timeouts.
+                    try:
+                        return await asyncio.wait_for(discovery, timeout=session_init_timeout)
+                    except TimeoutError:
+                        # Only our own bound is logged as "timed out": the
+                        # branch condition guarantees the value is not None, so
+                        # the %.1f format cannot fail. A TimeoutError raised by
+                        # discovery itself (e.g. an internal SDK timeout on the
+                        # opted-out path) falls through to the generic failure
+                        # handler below instead.
+                        logger.warning(
+                            "Skipping MCP server '%s' after tool discovery timed out (%.1fs)",
+                            server_name,
+                            session_init_timeout,
+                        )
+                        return []
+                return await discovery
             except Exception as e:
                 logger.warning(
                     f"Skipping MCP server '{server_name}' after tool discovery failed: {e}",
@@ -708,6 +783,7 @@ async def get_mcp_tools() -> list[BaseTool]:
                     tag_mcp_routing(tool, routing)
                 if transport == "stdio":
                     _timeout = server_cfg.tool_call_timeout if server_cfg else None
+                    _init_timeout = _resolve_session_init_timeout(server_cfg)
                     wrapped_tools.append(
                         _make_session_pool_tool(
                             tool,
@@ -715,6 +791,7 @@ async def get_mcp_tools() -> list[BaseTool]:
                             servers_config[source_name],
                             tool_interceptors,
                             tool_call_timeout=_timeout,
+                            session_init_timeout=_init_timeout,
                             tool_name_prefix=tool_name_prefix,
                         )
                     )

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -36,6 +36,7 @@ from langgraph.types import Overwrite
 from deerflow.agents.goal_state import GoalEvaluation, GoalState
 from deerflow.config.app_config import AppConfig
 from deerflow.config.database_config import CheckpointChannelMode
+from deerflow.constants import TOOL_RESULTS_DIRNAME
 from deerflow.runtime.checkpoint_mode import (
     aensure_checkpoint_mode_compatible,
     inject_checkpoint_mode,
@@ -205,17 +206,36 @@ def _delivery_error(content: dict[str, Any]) -> str | None:
     return _DELIVERY_INCOMPLETE_ERROR
 
 
+def _workspace_excluded_dir_names(app_config: AppConfig | None) -> frozenset[str]:
+    """Directory names workspace snapshots must skip for this deployment.
+
+    The tool-output budget middleware externalizes oversized tool outputs into
+    a storage subdir under outputs (default ``.tool-results``). Those files are
+    process feedback referenced from the budget preview via ``read_file``, not
+    deliverables: counting them as produced artifacts would fail run delivery
+    verification for any run that externalized a tool output without also
+    presenting a real artifact. The default name is excluded by the scanner
+    itself; a custom ``tool_output.storage_subdir`` (a single-segment name,
+    enforced by ``ToolOutputConfig`` so the scanner's dir-name pruning always
+    matches) is threaded through the snapshot capture here so before/after
+    diffs stay consistent.
+    """
+    storage_subdir = app_config.tool_output.storage_subdir if app_config is not None else TOOL_RESULTS_DIRNAME
+    return frozenset({storage_subdir})
+
+
 async def _produced_output_paths(
     before: WorkspaceSnapshot | None,
     *,
     thread_id: str,
     user_id: str | None,
+    extra_excluded_dir_names: frozenset[str] | None = None,
 ) -> list[str]:
     """Detect regular output files created or modified by this run."""
     if before is None:
         return []
     try:
-        after = await capture_workspace_snapshot(thread_id, user_id=user_id, include_text=False)
+        after = await capture_workspace_snapshot(thread_id, user_id=user_id, include_text=False, extra_excluded_dir_names=extra_excluded_dir_names)
         return get_changed_output_paths(before, after)
     except Exception:
         logger.warning("Could not detect produced output artifacts for run thread %s", thread_id, exc_info=True)
@@ -547,6 +567,7 @@ async def run_agent(
     pre_run_checkpoint_id: str | None = None
     pre_run_workspace_snapshot: WorkspaceSnapshot | None = None
     workspace_changes_user_id: str | None = None
+    workspace_excluded_dir_names: frozenset[str] | None = None
     snapshot_capture_failed = False
     llm_error_fallback_message: str | None = None
     checkpoint_rollback_completed = False
@@ -700,10 +721,15 @@ async def run_agent(
 
         if event_store is not None:
             workspace_changes_user_id = get_effective_user_id()
+            # Resolved once per run so the pre-run snapshot, the post-run
+            # delivery scan, and the workspace-changes scan all agree on the
+            # same exclusion set.
+            workspace_excluded_dir_names = _workspace_excluded_dir_names(ctx.app_config)
             try:
                 pre_run_workspace_snapshot = await capture_workspace_snapshot(
                     thread_id,
                     user_id=workspace_changes_user_id,
+                    extra_excluded_dir_names=workspace_excluded_dir_names,
                 )
             except Exception:
                 logger.warning("Could not capture pre-run workspace snapshot for run %s", run_id, exc_info=True)
@@ -1007,6 +1033,7 @@ async def run_agent(
                 pre_run_workspace_snapshot,
                 thread_id=thread_id,
                 user_id=workspace_changes_user_id,
+                extra_excluded_dir_names=workspace_excluded_dir_names,
             )
             delivery_content = _delivery_content_with_outputs(
                 journal.get_delivery_content() if journal is not None else _empty_delivery_content(),
@@ -1092,6 +1119,7 @@ async def run_agent(
                     run_id,
                     pre_run_workspace_snapshot,
                     user_id=workspace_changes_user_id,
+                    extra_excluded_dir_names=workspace_excluded_dir_names,
                 )
             except Exception:
                 logger.warning("Failed to record workspace changes for run %s", run_id, exc_info=True)
@@ -1113,6 +1141,7 @@ async def run_agent(
                         pre_run_workspace_snapshot,
                         thread_id=thread_id,
                         user_id=workspace_changes_user_id,
+                        extra_excluded_dir_names=workspace_excluded_dir_names,
                     )
                 delivery_content = _delivery_content_with_outputs(journal.get_delivery_content(), produced_output_paths)
             receipt_persisted = await _persist_delivery_receipt(

backend/packages/harness/deerflow/workspace_changes/recorder.py

@@ -81,6 +81,7 @@ async def capture_workspace_snapshot(
     user_id: str | None = None,
     limits: WorkspaceChangeLimits | None = None,
     include_text: bool = True,
+    extra_excluded_dir_names: frozenset[str] | None = None,
 ) -> WorkspaceSnapshot:
     # `_prepare_capture` creates the text cache dir inside the worker, so the
     # handoff must be cancellation-safe: if the run is cancelled after mkdtemp
@@ -110,6 +111,7 @@ async def capture_workspace_snapshot(
             limits=limits,
             include_text=include_text,
             text_cache_dir=text_cache_dir,
+            extra_excluded_dir_names=extra_excluded_dir_names,
         )
     except Exception:
         if text_cache_dir is not None:
@@ -125,6 +127,7 @@ async def record_workspace_changes(
     *,
     user_id: str | None = None,
     limits: WorkspaceChangeLimits | None = None,
+    extra_excluded_dir_names: frozenset[str] | None = None,
 ) -> dict | None:
     try:
         roots = await asyncio.to_thread(build_thread_workspace_roots, thread_id, user_id=user_id)
@@ -133,6 +136,7 @@ async def record_workspace_changes(
             roots,
             limits=limits,
             include_text=False,
+            extra_excluded_dir_names=extra_excluded_dir_names,
         )
         changed_paths = get_changed_paths(before, after_metadata)
         after = await asyncio.to_thread(
@@ -141,6 +145,7 @@ async def record_workspace_changes(
             limits=limits,
             include_text=True,
             text_paths=changed_paths,
+            extra_excluded_dir_names=extra_excluded_dir_names,
         )
         result = compare_snapshots(before, after, limits=limits)
         if not result.has_changes():

backend/packages/harness/deerflow/workspace_changes/scanner.py

@@ -6,7 +6,7 @@ import os
 from codecs import BOM_UTF16_BE, BOM_UTF16_LE, getincrementaldecoder
 from pathlib import Path
 
-from deerflow.constants import BROWSER_FRAMES_DIRNAME
+from deerflow.constants import BROWSER_FRAMES_DIRNAME, TOOL_RESULTS_DIRNAME
 
 from .types import (
     DiffUnavailableReason,
@@ -27,6 +27,14 @@ EXCLUDED_DIR_NAMES = {
     # the browser panel + inline thumbnails, not workspace deliverables. Shared
     # constant with the browser tools so the name cannot drift out of sync.
     BROWSER_FRAMES_DIRNAME,
+    # Externalized oversized tool outputs (the tool-output budget middleware's
+    # default storage_subdir): process feedback the model reads back via
+    # read_file, not workspace deliverables — same intent as the browser frames
+    # exclusion above. Without this, a run that externalizes any tool output
+    # would trip run delivery verification (produced output never presented)
+    # and fail as an error. Custom storage_subdir values are passed through
+    # ``extra_excluded_dir_names`` instead.
+    TOOL_RESULTS_DIRNAME,
     "__pycache__",
     "build",
     "dist",
@@ -102,11 +110,19 @@ def scan_workspace_roots(
     include_text: bool = True,
     text_paths: set[str] | None = None,
     text_cache_dir: Path | None = None,
+    extra_excluded_dir_names: frozenset[str] | None = None,
 ) -> WorkspaceSnapshot:
     resolved_limits = limits or WorkspaceChangeLimits()
     cache_dir = Path(text_cache_dir) if text_cache_dir is not None else None
     if cache_dir is not None:
         cache_dir.mkdir(parents=True, exist_ok=True)
+    # Operator-customized tool_output.storage_subdir values arrive here; the
+    # default name is already part of EXCLUDED_DIR_NAMES, so merging is safe.
+    # Only single-segment directory names are meaningful: os.walk yields
+    # one-segment dirnames, so a nested value like "cache/tool-results" would
+    # never match. ToolOutputConfig enforces the single-segment contract, so a
+    # multi-segment value is a caller error, not a silent no-op.
+    excluded_dir_names = EXCLUDED_DIR_NAMES | extra_excluded_dir_names if extra_excluded_dir_names else EXCLUDED_DIR_NAMES
     files: dict[str, FileSnapshot] = {}
     scanned = 0
     truncated = False
@@ -116,7 +132,7 @@ def scan_workspace_roots(
             continue
 
         for dirpath, dirnames, filenames in os.walk(root.host_path, followlinks=False):
-            dirnames[:] = [dirname for dirname in dirnames if dirname not in EXCLUDED_DIR_NAMES and not (Path(dirpath) / dirname).is_symlink()]
+            dirnames[:] = [dirname for dirname in dirnames if dirname not in excluded_dir_names and not (Path(dirpath) / dirname).is_symlink()]
             for filename in sorted(filenames):
                 if scanned >= resolved_limits.max_scanned_files:
                     truncated = True

backend/tests/test_mcp_session_pool.py

@@ -1730,7 +1730,7 @@ async def test_mcp_tools_routed_to_source_server_with_prefix_overlap():
 
     routed: list[tuple[str, str]] = []
 
-    def fake_wrap(tool, server_name, connection, interceptors, tool_call_timeout=None, tool_name_prefix=True):
+    def fake_wrap(tool, server_name, connection, interceptors, tool_call_timeout=None, session_init_timeout=None, tool_name_prefix=True):
         routed.append((tool.name, server_name))
         return tool
 

backend/tests/test_mcp_session_timeouts.py

@@ -0,0 +1,244 @@
+"""Timeout coverage for MCP server bring-up.
+
+``tool_call_timeout`` only bounds ``session.call_tool()``. Discovery
+(subprocess spawn + initialize + tools/list) and persistent-session
+initialization have no bound on their own, so a hung stdio server would block
+agent construction forever. These tests pin the ``session_init_timeout`` bound
+on both stages and the per-server independence of the discovery timeout.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from langchain_core.tools import StructuredTool
+from pydantic import BaseModel, Field
+
+from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig
+from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT
+from deerflow.mcp.tools import _make_session_pool_tool, get_mcp_tools
+
+
+class _Args(BaseModel):
+    query: str = Field(..., description="query")
+
+
+def _tool(name: str) -> StructuredTool:
+    async def _call(query: str) -> str:
+        return query
+
+    return StructuredTool(
+        name=name,
+        description="Search",
+        args_schema=_Args,
+        coroutine=_call,
+    )
+
+
+def test_session_init_timeout_defaults_to_shared_constant() -> None:
+    assert McpServerConfig().session_init_timeout == DEFAULT_MCP_SESSION_INIT_TIMEOUT
+    assert McpServerConfig(session_init_timeout=None).session_init_timeout is None
+
+
+@pytest.mark.asyncio
+async def test_discovery_timeout_skips_hung_server_without_blocking_healthy_server() -> None:
+    """A server whose discovery hangs must time out and be skipped, while a
+    healthy server still contributes its tools."""
+    extensions_config = ExtensionsConfig.model_validate(
+        {
+            "mcpServers": {
+                "slow_server": {
+                    "type": "stdio",
+                    "command": "uvx",
+                    "args": ["slow-mcp"],
+                    "session_init_timeout": 0.05,
+                },
+                "fast_server": {
+                    "type": "stdio",
+                    "command": "uvx",
+                    "args": ["fast-mcp"],
+                    "session_init_timeout": 1.0,
+                },
+            }
+        }
+    )
+    servers_config = {
+        "slow_server": {"transport": "stdio", "command": "uvx", "args": ["slow-mcp"]},
+        "fast_server": {"transport": "stdio", "command": "uvx", "args": ["fast-mcp"]},
+    }
+
+    class FakeClient:
+        def __init__(
+            self,
+            connections,
+            *,
+            callbacks=None,
+            tool_interceptors=None,
+            tool_name_prefix=False,
+        ) -> None:
+            self.connections = connections
+            self.callbacks = callbacks
+            self.tool_interceptors = tool_interceptors or []
+            self.tool_name_prefix = tool_name_prefix
+
+        async def get_tools(self, *, server_name=None):
+            if server_name == "slow_server":
+                await asyncio.sleep(60)  # hung discovery
+            # The real adapter returns server-prefixed tool names when
+            # tool_name_prefix=True.
+            return [_tool("fast_server_fast_search")]
+
+    with (
+        patch("deerflow.mcp.tools.ExtensionsConfig.from_file", return_value=extensions_config),
+        patch("deerflow.mcp.tools.build_servers_config", return_value=servers_config),
+        patch("deerflow.mcp.tools.get_initial_oauth_headers", new_callable=AsyncMock, return_value={}),
+        patch("deerflow.mcp.tools.build_oauth_tool_interceptor", return_value=None),
+        patch("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient),
+        patch("langchain_mcp_adapters.tools.load_mcp_tools", new_callable=AsyncMock),
+        patch("deerflow.mcp.tools._make_session_pool_tool", side_effect=lambda tool, *_args, **_kwargs: tool),
+    ):
+        # Without the discovery timeout the slow server would hang the call past
+        # the 5s bound and this test would fail with TimeoutError.
+        tools = await asyncio.wait_for(get_mcp_tools(), timeout=5)
+
+    assert [tool.name for tool in tools] == ["fast_server_fast_search"]
+
+
+@pytest.mark.asyncio
+async def test_session_init_timeout_raises_when_session_creation_hangs(tmp_path, caplog) -> None:
+    """A server that never finishes initialize() must not block the tool call,
+    and the timeout must be visible in logs at the same level as discovery
+    timeouts so operators can diagnose hung MCP sessions."""
+    mock_pool = MagicMock()
+
+    async def hanging_get_session(*_args, **_kwargs) -> None:
+        await asyncio.sleep(60)
+
+    mock_pool.get_session = hanging_get_session
+
+    with (
+        patch("deerflow.mcp.tools.get_session_pool", return_value=mock_pool),
+        patch("deerflow.mcp.tools.get_paths", return_value=MagicMock()),
+        patch(
+            "deerflow.mcp.tools._prepare_stdio_workspace",
+            return_value=(tmp_path, tmp_path / "tmp", {}),
+        ),
+        caplog.at_level(logging.WARNING, logger="deerflow.mcp.tools"),
+    ):
+        wrapped = _make_session_pool_tool(
+            _tool("github_search"),
+            "github",
+            {"transport": "stdio", "command": "mcp-server", "args": []},
+            session_init_timeout=0.05,
+            tool_name_prefix=False,
+        )
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        with pytest.raises(TimeoutError):
+            await wrapped.coroutine(query="repositories")
+        # Bounds the regression: the timeout must fire promptly, not wait on the
+        # hung session.
+        assert loop.time() - start < 1.0
+
+    timeout_warnings = [record for record in caplog.records if record.levelno == logging.WARNING and "timed out" in record.getMessage()]
+    assert timeout_warnings, "session-init timeout must be logged like discovery timeouts"
+    assert "github" in timeout_warnings[0].getMessage()
+
+
+@pytest.mark.asyncio
+async def test_discovery_timeout_from_sdk_with_opt_out_is_reported_without_logging_error(caplog) -> None:
+    """With session_init_timeout opted out (None), a TimeoutError raised by
+    discovery itself (e.g. an internal timeout inside the MCP SDK) must still
+    be reported gracefully. The skip must go through the generic failure path —
+    never through the "timed out (%.1fs)" format with a None value, which
+    would raise inside the logging module and silently drop the warning."""
+    extensions_config = ExtensionsConfig.model_validate(
+        {
+            "mcpServers": {
+                "flaky_server": {
+                    "type": "stdio",
+                    "command": "uvx",
+                    "args": ["flaky-mcp"],
+                    "session_init_timeout": None,
+                },
+            }
+        }
+    )
+    servers_config = {
+        "flaky_server": {"transport": "stdio", "command": "uvx", "args": ["flaky-mcp"]},
+    }
+
+    class FakeClient:
+        def __init__(
+            self,
+            connections,
+            *,
+            callbacks=None,
+            tool_interceptors=None,
+            tool_name_prefix=False,
+        ) -> None:
+            self.callbacks = callbacks
+            self.tool_interceptors = tool_interceptors or []
+            self.tool_name_prefix = tool_name_prefix
+
+        async def get_tools(self, *, server_name=None):
+            raise TimeoutError("internal SDK timeout")
+
+    with (
+        patch("deerflow.mcp.tools.ExtensionsConfig.from_file", return_value=extensions_config),
+        patch("deerflow.mcp.tools.build_servers_config", return_value=servers_config),
+        patch("deerflow.mcp.tools.get_initial_oauth_headers", new_callable=AsyncMock, return_value={}),
+        patch("deerflow.mcp.tools.build_oauth_tool_interceptor", return_value=None),
+        patch("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient),
+        patch("langchain_mcp_adapters.tools.load_mcp_tools", new_callable=AsyncMock),
+        caplog.at_level(logging.WARNING, logger="deerflow.mcp.tools"),
+    ):
+        tools = await get_mcp_tools()
+
+    assert tools == []
+    # getMessage() on every captured record must not raise: pre-fix, the only
+    # record for this server was the broken "timed out (%.1fs)" % None format.
+    assert any("tool discovery failed" in record.getMessage() for record in caplog.records)
+    assert not any("timed out" in record.getMessage() for record in caplog.records)
+
+
+def test_gateway_response_model_session_init_timeout_default_matches_runtime_config() -> None:
+    """A server created via PUT /api/mcp/config without session_init_timeout
+    must get the same bring-up timeout as one created in the config file —
+    the response model's default feeds model_dump() into the persisted config."""
+    from app.gateway.routers.mcp import McpServerConfigResponse
+
+    assert McpServerConfigResponse.model_validate({}).session_init_timeout == DEFAULT_MCP_SESSION_INIT_TIMEOUT
+    # An explicit null stays an explicit opt-out (no timeout).
+    assert McpServerConfigResponse.model_validate({"session_init_timeout": None}).session_init_timeout is None
+
+
+@pytest.mark.asyncio
+async def test_session_init_timeout_does_not_block_fast_session(tmp_path) -> None:
+    """A promptly-initialized session still completes the tool call."""
+    mock_session = AsyncMock()
+    mock_session.call_tool = AsyncMock(return_value=MagicMock(content=[], isError=False, structuredContent=None))
+    mock_pool = MagicMock()
+    mock_pool.get_session = AsyncMock(return_value=mock_session)
+
+    with (
+        patch("deerflow.mcp.tools.get_session_pool", return_value=mock_pool),
+        patch("deerflow.mcp.tools.get_paths", return_value=MagicMock()),
+        patch(
+            "deerflow.mcp.tools._prepare_stdio_workspace",
+            return_value=(tmp_path, tmp_path / "tmp", {}),
+        ),
+    ):
+        wrapped = _make_session_pool_tool(
+            _tool("github_search"),
+            "github",
+            {"transport": "stdio", "command": "mcp-server", "args": []},
+            session_init_timeout=5.0,
+            tool_name_prefix=False,
+        )
+        await wrapped.coroutine(query="repositories")
+
+    mock_session.call_tool.assert_awaited_once_with("github_search", {"query": "repositories"})

backend/tests/test_run_worker_delivery.py

@@ -9,11 +9,16 @@ import pytest
 from langchain_core.messages import AIMessage, ToolMessage
 from langgraph.types import Command
 
+from deerflow.config.app_config import AppConfig
+from deerflow.config.paths import Paths
+from deerflow.config.sandbox_config import SandboxConfig
+from deerflow.config.tool_output_config import ToolOutputConfig
 from deerflow.runtime.events.store.memory import MemoryRunEventStore
 from deerflow.runtime.runs.manager import RunManager
 from deerflow.runtime.runs.schemas import RunStatus
 from deerflow.runtime.runs.store.memory import MemoryRunStore
 from deerflow.runtime.runs.worker import RunContext, _delivery_content_with_outputs, run_agent
+from deerflow.runtime.user_context import get_effective_user_id
 
 
 def _make_bridge():
@@ -215,6 +220,75 @@ async def test_changed_outputs_fail_closed_when_not_presented(monkeypatch):
     assert record.stop_reason is None
 
 
+@pytest.mark.anyio
+async def test_externalized_tool_results_do_not_trigger_delivery_verification(tmp_path, monkeypatch):
+    """Oversized tool outputs externalized under outputs/.tool-results/ are
+    process feedback for the model, not deliverables: a run that only produced
+    those files must succeed without any present_files call."""
+    paths = Paths(base_dir=tmp_path)
+    monkeypatch.setattr("deerflow.workspace_changes.recorder.get_paths", lambda: paths)
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+
+    class ExternalizingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            # Simulates ToolOutputBudgetMiddleware persisting an oversized tool
+            # output mid-run (default storage_subdir is ".tool-results").
+            tool_results = paths.sandbox_outputs_dir("thread-1", user_id=get_effective_user_id()) / ".tool-results"
+            tool_results.mkdir(parents=True, exist_ok=True)
+            (tool_results / "bash-abcdef123456.log").write_text("x" * 20000, encoding="utf-8")
+            yield {"messages": [AIMessage(content="Here is the answer.")]}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: ExternalizingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.success
+
+
+@pytest.mark.anyio
+async def test_custom_tool_output_storage_subdir_does_not_trigger_delivery_verification(tmp_path, monkeypatch):
+    """A custom tool_output.storage_subdir is honoured by the exclusion, not
+    only the default .tool-results name."""
+    paths = Paths(base_dir=tmp_path)
+    monkeypatch.setattr("deerflow.workspace_changes.recorder.get_paths", lambda: paths)
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    app_config = AppConfig(sandbox=SandboxConfig(use="test"), tool_output=ToolOutputConfig(storage_subdir="tool-output-cache"))
+
+    class ExternalizingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            cache = paths.sandbox_outputs_dir("thread-1", user_id=get_effective_user_id()) / "tool-output-cache"
+            cache.mkdir(parents=True, exist_ok=True)
+            (cache / "web_fetch-abcdef123456.log").write_text("y" * 20000, encoding="utf-8")
+            yield {"messages": [AIMessage(content="Here is the answer.")]}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store, app_config=app_config),
+        agent_factory=lambda *, config: ExternalizingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.success
+
+
 @pytest.mark.anyio
 async def test_changed_outputs_succeed_when_one_of_multiple_outputs_is_presented(monkeypatch):
     run_manager = RunManager()

backend/tests/test_tool_output_budget_middleware.py

@@ -242,6 +242,45 @@ class TestExternalizePathTraversal:
         assert path is None
 
 
+class TestStorageSubdirConfig:
+    """storage_subdir must be a single path segment.
+
+    The workspace-changes scanner prunes by directory name during os.walk,
+    which yields one-segment dirnames — a nested value like
+    ``cache/tool-results`` would silently never match the exclusion and its
+    files would be counted as produced artifacts again. Rejecting it at config
+    time keeps the dir-name-based exclusion sound.
+    """
+
+    def test_default_is_single_segment(self):
+        assert ToolOutputConfig().storage_subdir == ".tool-results"
+
+    def test_single_segment_custom_accepted(self):
+        assert ToolOutputConfig(storage_subdir="tool-output-cache").storage_subdir == "tool-output-cache"
+
+    def test_nested_storage_subdir_rejected(self):
+        with pytest.raises(ValueError):
+            ToolOutputConfig(storage_subdir="cache/tool-results")
+
+    def test_windows_separator_storage_subdir_rejected(self):
+        with pytest.raises(ValueError):
+            ToolOutputConfig(storage_subdir="cache\\tool-results")
+
+    def test_absolute_storage_subdir_rejected(self):
+        with pytest.raises(ValueError):
+            ToolOutputConfig(storage_subdir="/abs/path")
+
+    def test_dot_and_dotdot_storage_subdir_rejected(self):
+        with pytest.raises(ValueError):
+            ToolOutputConfig(storage_subdir=".")
+        with pytest.raises(ValueError):
+            ToolOutputConfig(storage_subdir="..")
+
+    def test_empty_storage_subdir_rejected(self):
+        with pytest.raises(ValueError):
+            ToolOutputConfig(storage_subdir="")
+
+
 class TestNeedsBudget:
     def test_small_output_does_not_need_budget(self):
         config = ToolOutputConfig(externalize_min_chars=1000)

backend/tests/test_workspace_changes.py

@@ -261,6 +261,48 @@ def test_scan_workspace_roots_skips_browser_frames(tmp_path):
     assert "/mnt/user-data/outputs/.browser-frames/browser-navigate-1.png" not in snapshot.files
 
 
+def test_scan_workspace_roots_skips_externalized_tool_results(tmp_path):
+    roots = _roots(tmp_path)
+    outputs = roots[1].host_path
+    (outputs / "report.md").write_text("keep", encoding="utf-8")
+    tool_results = outputs / ".tool-results"
+    tool_results.mkdir()
+    (tool_results / "bash-abcdef123456.log").write_text("oversized tool output", encoding="utf-8")
+
+    snapshot = scan_workspace_roots(roots)
+
+    assert "/mnt/user-data/outputs/report.md" in snapshot.files
+    assert "/mnt/user-data/outputs/.tool-results/bash-abcdef123456.log" not in snapshot.files
+
+
+def test_scan_workspace_roots_skips_extra_excluded_dir_names(tmp_path):
+    roots = _roots(tmp_path)
+    outputs = roots[1].host_path
+    (outputs / "report.md").write_text("keep", encoding="utf-8")
+    custom = outputs / "custom-tool-results"
+    custom.mkdir()
+    (custom / "bash-abcdef123456.log").write_text("oversized tool output", encoding="utf-8")
+
+    snapshot = scan_workspace_roots(roots, extra_excluded_dir_names=frozenset({"custom-tool-results"}))
+
+    assert "/mnt/user-data/outputs/report.md" in snapshot.files
+    assert "/mnt/user-data/outputs/custom-tool-results/bash-abcdef123456.log" not in snapshot.files
+
+
+def test_get_changed_output_paths_ignores_externalized_tool_results(tmp_path):
+    roots = _roots(tmp_path)
+    outputs = roots[1].host_path
+    before = scan_workspace_roots(roots)
+
+    tool_results = outputs / ".tool-results"
+    tool_results.mkdir()
+    (tool_results / "web_fetch-abcdef123456.log").write_text("x" * 20000, encoding="utf-8")
+    (outputs / "report.md").write_text("deliverable", encoding="utf-8")
+    after = scan_workspace_roots(roots)
+
+    assert get_changed_output_paths(before, after) == ["/mnt/user-data/outputs/report.md"]
+
+
 def test_scan_workspace_roots_can_skip_text_loading(tmp_path):
     roots = _roots(tmp_path)
     workspace = roots[0].host_path

extensions_config.example.json

@@ -16,6 +16,7 @@
         "GITHUB_TOKEN": "$GITHUB_TOKEN"
       },
       "tool_name_prefix": true,
+      "session_init_timeout": 60,
       "tool_call_timeout": 60,
       "description": "GitHub MCP server for repository operations"
     },