fix(copilot): unified MCP Write tool to prevent truncation data loss

Replace the CLI's built-in Write tool with a unified MCP Write tool that
detects and handles output-token truncation gracefully instead of losing
user work with opaque "'file_path' is a required property" errors.

The new tool:
- Works in both E2B and non-E2B modes
- Detects partial truncation (content present but file_path missing)
- Detects complete truncation (empty args) with actionable guidance
- Warns on large content (>50K chars) that succeeded
- Places file_path first in schema to survive truncation better
- Validates paths stay within the SDK working directory

Also blocks the CLI built-in Write via SDK_DISALLOWED_TOOLS and
strengthens the system prompt's large-file writing guidance.

Fixes production truncation bug reported in sessions:
  1a0ef47f-9711-41d2-91b8-df9dff9ecfc6
  2bb9fb0d-05bd-4194-8f3f-88fbe3d8b965
  8226d82e-bf9c-48e3-826f-80048d0fb7b4

autogpt_platform/backend/backend/copilot/prompting.py

@@ -75,11 +75,12 @@ Example — committing an image file to GitHub:
 }}
 ```
 
-### Writing large files — CRITICAL
-**Never write an entire large document in a single tool call.**  When the
-content you want to write exceeds ~2000 words the tool call's output token
-limit will silently truncate the arguments, producing an empty `{{}}` input
-that fails repeatedly.
+### Writing large files — CRITICAL (causes production failures)
+**NEVER write an entire large document in a single tool call.**  When the
+content you want to write exceeds ~2000 words the API output-token limit
+will silently truncate the tool call arguments mid-JSON, losing all content
+and producing an opaque error.  This is unrecoverable — the user's work is
+lost and retrying with the same approach fails in an infinite loop.
 
 **Preferred: compose from file references.**  If the data is already in
 files (tool outputs, workspace files), compose the report in one call

autogpt_platform/backend/backend/copilot/sdk/file_tools.py

@@ -0,0 +1,172 @@
+"""Unified MCP Write tool that works in both E2B and non-E2B modes.
+
+Replaces the CLI's built-in Write tool, which has no defence against output-token
+truncation.  When the LLM generates a very large ``content`` argument the API
+truncates the response mid-JSON and Ajv rejects it with the opaque
+"'file_path' is a required property" error, losing the user's work.
+
+This MCP tool:
+- Detects partial truncation (content present but file_path missing)
+- Detects complete truncation (empty args)
+- Warns on large content that succeeded (>50K chars)
+- In non-E2B mode: writes to the SDK working directory
+- In E2B mode: delegates to the E2B sandbox write handler
+
+The JSON schema places ``file_path`` FIRST so that truncation is more likely
+to preserve the path (the API serialises properties in schema order).
+"""
+
+import json
+import logging
+import os
+from typing import Any, Callable
+
+from backend.copilot.context import (
+    get_sdk_cwd,
+    is_allowed_local_path,
+)
+
+logger = logging.getLogger(__name__)
+
+# Inline content above this threshold triggers a warning — it survived this
+# time but is dangerously close to the API output-token truncation limit.
+_LARGE_CONTENT_WARN_CHARS = 50_000
+
+
+def _mcp(text: str, *, error: bool = False) -> dict[str, Any]:
+    if error:
+        text = json.dumps({"error": text, "type": "error"})
+    return {"content": [{"type": "text", "text": text}], "isError": error}
+
+
+async def _handle_write_non_e2b(args: dict[str, Any]) -> dict[str, Any]:
+    """Write content to a file in the SDK working directory (non-E2B mode)."""
+    file_path: str = args.get("file_path", "")
+    content: str = args.get("content", "")
+
+    if not file_path:
+        if content:
+            return _mcp(
+                "Your Write call was truncated (file_path missing but content "
+                "was present). The content was too large for a single tool call. "
+                "Write in chunks: use bash_exec with "
+                "'cat > file << \"EOF\"\\n...\\nEOF' for the first section, "
+                "'cat >> file << \"EOF\"\\n...\\nEOF' to append subsequent "
+                "sections, then reference the file with "
+                "@@agptfile:/path/to/file if needed.",
+                error=True,
+            )
+        return _mcp(
+            "Your Write call had empty arguments — this means your previous "
+            "response was too long and the tool call was truncated by the API. "
+            "Break your work into smaller steps. For large content, write "
+            "section-by-section using bash_exec with "
+            "'cat > file << \"EOF\"\\n...\\nEOF' and "
+            "'cat >> file << \"EOF\"\\n...\\nEOF'.",
+            error=True,
+        )
+
+    sdk_cwd = get_sdk_cwd()
+    if not sdk_cwd:
+        return _mcp("No SDK working directory available", error=True)
+
+    # Resolve relative paths against SDK working directory
+    if not os.path.isabs(file_path):
+        resolved = os.path.normpath(os.path.join(sdk_cwd, file_path))
+    else:
+        resolved = os.path.normpath(file_path)
+
+    # Validate path stays within allowed directories
+    if not is_allowed_local_path(resolved, sdk_cwd):
+        return _mcp(
+            f"Path must be within the working directory ({sdk_cwd}): {file_path}",
+            error=True,
+        )
+
+    try:
+        parent = os.path.dirname(resolved)
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        with open(resolved, "w", encoding="utf-8") as f:
+            f.write(content)
+    except Exception as exc:
+        return _mcp(f"Failed to write {resolved}: {exc}", error=True)
+
+    msg = f"Successfully wrote to {resolved}"
+    if len(content) > _LARGE_CONTENT_WARN_CHARS:
+        logger.warning(
+            "[Write] large inline content (%d chars) for %s",
+            len(content),
+            resolved,
+        )
+        msg += (
+            f"\n\nWARNING: The content was very large ({len(content)} chars). "
+            "Next time, write large files in sections using bash_exec with "
+            "'cat > file << EOF ... EOF' and 'cat >> file << EOF ... EOF' "
+            "to avoid output-token truncation."
+        )
+    return _mcp(msg)
+
+
+async def _handle_write_e2b(args: dict[str, Any]) -> dict[str, Any]:
+    """Write content to a file, delegating to the E2B sandbox."""
+    from .e2b_file_tools import _handle_write_file
+
+    file_path: str = args.get("file_path", "")
+    content: str = args.get("content", "")
+
+    if not file_path:
+        if content:
+            return _mcp(
+                "Your Write call was truncated (file_path missing but content "
+                "was present). The content was too large for a single tool call. "
+                "Write in chunks: use bash_exec with "
+                "'cat > file << \"EOF\"\\n...\\nEOF' for the first section, "
+                "'cat >> file << \"EOF\"\\n...\\nEOF' to append subsequent "
+                "sections, then reference the file with "
+                "@@agptfile:/path/to/file if needed.",
+                error=True,
+            )
+        return _mcp(
+            "Your Write call had empty arguments — this means your previous "
+            "response was too long and the tool call was truncated by the API. "
+            "Break your work into smaller steps. For large content, write "
+            "section-by-section using bash_exec with "
+            "'cat > file << \"EOF\"\\n...\\nEOF' and "
+            "'cat >> file << \"EOF\"\\n...\\nEOF'.",
+            error=True,
+        )
+
+    return await _handle_write_file(args)
+
+
+def get_write_tool_handler(*, use_e2b: bool) -> Callable[..., Any]:
+    """Return the appropriate Write handler for the current execution mode."""
+    if use_e2b:
+        return _handle_write_e2b
+    return _handle_write_non_e2b
+
+
+WRITE_TOOL_NAME = "Write"
+WRITE_TOOL_DESCRIPTION = (
+    "Write or create a file. Parent directories are created automatically. "
+    "For large content (>2000 words), prefer writing in sections using "
+    "bash_exec with 'cat > file' and 'cat >> file' instead."
+)
+WRITE_TOOL_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "properties": {
+        "file_path": {
+            "type": "string",
+            "description": (
+                "The path to the file to write. "
+                "Relative paths are resolved against the working directory."
+            ),
+        },
+        "content": {
+            "type": "string",
+            "description": "The content to write to the file.",
+        },
+    },
+    "required": ["file_path", "content"],
+}

autogpt_platform/backend/backend/copilot/sdk/file_tools_test.py

@@ -0,0 +1,196 @@
+"""Tests for the unified MCP Write tool (file_tools.py).
+
+Covers: normal write, large content warning, partial truncation,
+complete truncation, path validation (no escape from working dir),
+E2B delegation, and CLI built-in Write disallowance.
+"""
+
+import os
+
+import pytest
+
+from backend.copilot.sdk.tool_adapter import SDK_DISALLOWED_TOOLS
+
+from .file_tools import (
+    _LARGE_CONTENT_WARN_CHARS,
+    WRITE_TOOL_NAME,
+    WRITE_TOOL_SCHEMA,
+    _handle_write_non_e2b,
+)
+
+
+@pytest.fixture
+def sdk_cwd(tmp_path, monkeypatch):
+    """Provide a temporary SDK working directory."""
+    cwd = str(tmp_path / "copilot-test-session")
+    os.makedirs(cwd, exist_ok=True)
+    monkeypatch.setattr("backend.copilot.sdk.file_tools.get_sdk_cwd", lambda: cwd)
+    # Patch is_allowed_local_path to allow paths under our tmp cwd
+
+    def _patched_is_allowed(path: str, cwd_arg: str | None = None) -> bool:
+        resolved = os.path.realpath(path)
+        norm_cwd = os.path.realpath(cwd)
+        return resolved == norm_cwd or resolved.startswith(norm_cwd + os.sep)
+
+    monkeypatch.setattr(
+        "backend.copilot.sdk.file_tools.is_allowed_local_path",
+        _patched_is_allowed,
+    )
+    return cwd
+
+
+# ---------------------------------------------------------------------------
+# Schema validation
+# ---------------------------------------------------------------------------
+
+
+class TestWriteToolSchema:
+    def test_file_path_is_first_property(self):
+        """file_path should be listed first in schema so truncation preserves it."""
+        props = list(WRITE_TOOL_SCHEMA["properties"].keys())
+        assert props[0] == "file_path"
+
+    def test_both_fields_required(self):
+        assert "file_path" in WRITE_TOOL_SCHEMA["required"]
+        assert "content" in WRITE_TOOL_SCHEMA["required"]
+
+
+# ---------------------------------------------------------------------------
+# Normal write
+# ---------------------------------------------------------------------------
+
+
+class TestNormalWrite:
+    @pytest.mark.asyncio
+    async def test_write_creates_file(self, sdk_cwd):
+        result = await _handle_write_non_e2b(
+            {"file_path": "hello.txt", "content": "Hello, world!"}
+        )
+        assert not result["isError"]
+        written = open(os.path.join(sdk_cwd, "hello.txt")).read()
+        assert written == "Hello, world!"
+
+    @pytest.mark.asyncio
+    async def test_write_creates_parent_dirs(self, sdk_cwd):
+        result = await _handle_write_non_e2b(
+            {"file_path": "sub/dir/file.py", "content": "print('hi')"}
+        )
+        assert not result["isError"]
+        assert os.path.isfile(os.path.join(sdk_cwd, "sub", "dir", "file.py"))
+
+    @pytest.mark.asyncio
+    async def test_write_absolute_path_within_cwd(self, sdk_cwd):
+        abs_path = os.path.join(sdk_cwd, "abs.txt")
+        result = await _handle_write_non_e2b(
+            {"file_path": abs_path, "content": "absolute"}
+        )
+        assert not result["isError"]
+        assert open(abs_path).read() == "absolute"
+
+    @pytest.mark.asyncio
+    async def test_success_message_contains_path(self, sdk_cwd):
+        result = await _handle_write_non_e2b({"file_path": "msg.txt", "content": "ok"})
+        text = result["content"][0]["text"]
+        assert "Successfully wrote" in text
+        assert "msg.txt" in text
+
+
+# ---------------------------------------------------------------------------
+# Large content warning
+# ---------------------------------------------------------------------------
+
+
+class TestLargeContentWarning:
+    @pytest.mark.asyncio
+    async def test_large_content_warns(self, sdk_cwd):
+        big_content = "x" * (_LARGE_CONTENT_WARN_CHARS + 1)
+        result = await _handle_write_non_e2b(
+            {"file_path": "big.txt", "content": big_content}
+        )
+        assert not result["isError"]
+        text = result["content"][0]["text"]
+        assert "WARNING" in text
+        assert "large" in text.lower()
+
+    @pytest.mark.asyncio
+    async def test_normal_content_no_warning(self, sdk_cwd):
+        result = await _handle_write_non_e2b(
+            {"file_path": "small.txt", "content": "small"}
+        )
+        text = result["content"][0]["text"]
+        assert "WARNING" not in text
+
+
+# ---------------------------------------------------------------------------
+# Truncation detection
+# ---------------------------------------------------------------------------
+
+
+class TestTruncationDetection:
+    @pytest.mark.asyncio
+    async def test_partial_truncation_content_no_path(self, sdk_cwd):
+        """Simulates API truncating file_path but preserving content."""
+        result = await _handle_write_non_e2b({"content": "some content here"})
+        assert result["isError"]
+        text = result["content"][0]["text"]
+        assert "truncated" in text.lower()
+        assert "file_path" in text.lower()
+
+    @pytest.mark.asyncio
+    async def test_complete_truncation_empty_args(self, sdk_cwd):
+        """Simulates API truncating to empty args {}."""
+        result = await _handle_write_non_e2b({})
+        assert result["isError"]
+        text = result["content"][0]["text"]
+        assert "truncated" in text.lower()
+        assert "smaller steps" in text.lower()
+
+    @pytest.mark.asyncio
+    async def test_empty_file_path_string(self, sdk_cwd):
+        """Empty string file_path should trigger truncation error."""
+        result = await _handle_write_non_e2b({"file_path": "", "content": "data"})
+        assert result["isError"]
+
+
+# ---------------------------------------------------------------------------
+# Path validation
+# ---------------------------------------------------------------------------
+
+
+class TestPathValidation:
+    @pytest.mark.asyncio
+    async def test_path_traversal_blocked(self, sdk_cwd):
+        result = await _handle_write_non_e2b(
+            {"file_path": "../../etc/passwd", "content": "evil"}
+        )
+        assert result["isError"]
+        text = result["content"][0]["text"]
+        assert "must be within" in text.lower()
+
+    @pytest.mark.asyncio
+    async def test_absolute_outside_cwd_blocked(self, sdk_cwd):
+        result = await _handle_write_non_e2b(
+            {"file_path": "/etc/passwd", "content": "evil"}
+        )
+        assert result["isError"]
+
+    @pytest.mark.asyncio
+    async def test_no_sdk_cwd_returns_error(self, monkeypatch):
+        monkeypatch.setattr("backend.copilot.sdk.file_tools.get_sdk_cwd", lambda: "")
+        result = await _handle_write_non_e2b({"file_path": "test.txt", "content": "hi"})
+        assert result["isError"]
+        text = result["content"][0]["text"]
+        assert "working directory" in text.lower()
+
+
+# ---------------------------------------------------------------------------
+# CLI built-in Write is disallowed
+# ---------------------------------------------------------------------------
+
+
+class TestCliBuiltinWriteDisallowed:
+    def test_write_in_disallowed_tools(self):
+        assert "Write" in SDK_DISALLOWED_TOOLS
+
+    def test_tool_name_is_write(self):
+        assert WRITE_TOOL_NAME == "Write"

autogpt_platform/backend/backend/copilot/sdk/tool_adapter.py

@@ -39,6 +39,12 @@ from backend.copilot.tools.base import BaseTool
 from backend.util.truncate import truncate
 
 from .e2b_file_tools import E2B_FILE_TOOL_NAMES, E2B_FILE_TOOLS, bridge_and_annotate
+from .file_tools import (
+    WRITE_TOOL_DESCRIPTION,
+    WRITE_TOOL_NAME,
+    WRITE_TOOL_SCHEMA,
+    get_write_tool_handler,
+)
 
 if TYPE_CHECKING:
     from e2b import AsyncSandbox
@@ -619,6 +625,22 @@ def create_copilot_mcp_server(*, use_e2b: bool = False):
             )(_make_truncating_wrapper(handler, name))
             sdk_tools.append(decorated)
 
+    # Unified Write tool — replaces the CLI's built-in Write which has no
+    # defence against output-token truncation.  Registered in both E2B and
+    # non-E2B modes so we always control validation and error messaging.
+    write_handler = get_write_tool_handler(use_e2b=use_e2b)
+    write_tool = tool(
+        WRITE_TOOL_NAME,
+        WRITE_TOOL_DESCRIPTION,
+        WRITE_TOOL_SCHEMA,
+        annotations=_PARALLEL_ANNOTATION,
+    )(
+        _make_truncating_wrapper(
+            write_handler, WRITE_TOOL_NAME, input_schema=WRITE_TOOL_SCHEMA
+        )
+    )
+    sdk_tools.append(write_tool)
+
     # Read tool for SDK-truncated tool results (always needed, read-only).
     read_tool = tool(
         _READ_TOOL_NAME,
@@ -655,10 +677,17 @@ _SDK_BUILTIN_TOOLS = [*_SDK_BUILTIN_FILE_TOOLS, *_SDK_BUILTIN_ALWAYS]
 # WebFetch: SSRF risk — can reach internal network (localhost, 10.x, etc.).
 #   Agent uses the SSRF-protected mcp__copilot__web_fetch tool instead.
 # AskUserQuestion: interactive CLI tool — no terminal in copilot context.
+# Write: the CLI's built-in Write tool has no defence against output-token
+#   truncation.  When the LLM generates a very large `content` argument the
+#   API truncates the response mid-JSON and Ajv rejects it with the opaque
+#   "'file_path' is a required property" error, losing the user's work.
+#   All writes go through our MCP Write tool (file_tools.py) where we
+#   control validation and return actionable guidance.
 SDK_DISALLOWED_TOOLS = [
     "Bash",
     "WebFetch",
     "AskUserQuestion",
+    "Write",
 ]
 
 # Tools that are blocked entirely in security hooks (defence-in-depth).
@@ -697,6 +726,7 @@ DANGEROUS_PATTERNS = [
 # Static tool name list for the non-E2B case (backward compatibility).
 COPILOT_TOOL_NAMES = [
     *[f"{MCP_TOOL_PREFIX}{name}" for name in TOOL_REGISTRY.keys()],
+    f"{MCP_TOOL_PREFIX}{WRITE_TOOL_NAME}",
     f"{MCP_TOOL_PREFIX}{_READ_TOOL_NAME}",
     *_SDK_BUILTIN_TOOLS,
 ]
@@ -713,6 +743,7 @@ def get_copilot_tool_names(*, use_e2b: bool = False) -> list[str]:
 
     return [
         *[f"{MCP_TOOL_PREFIX}{name}" for name in TOOL_REGISTRY.keys()],
+        f"{MCP_TOOL_PREFIX}{WRITE_TOOL_NAME}",
         f"{MCP_TOOL_PREFIX}{_READ_TOOL_NAME}",
         *[f"{MCP_TOOL_PREFIX}{name}" for name in E2B_FILE_TOOL_NAMES],
         *_SDK_BUILTIN_ALWAYS,