fix(copilot): unified MCP file tools (Read/Write/Edit) to prevent truncation data loss

autogpt_platform/backend/backend/copilot/prompting.py

@@ -75,11 +75,12 @@
 }}
 ```
 
-### 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,162 @@
+"""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}
+
+
+_PARTIAL_TRUNCATION_MSG = (
+    "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."
+)
+
+_COMPLETE_TRUNCATION_MSG = (
+    "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'."
+)
+
+
+def _check_truncation(file_path: str, content: str) -> dict[str, Any] | None:
+    """Return an error response if the args look truncated, else ``None``."""
+    if not file_path:
+        if content:
+            return _mcp(_PARTIAL_TRUNCATION_MSG, error=True)
+        return _mcp(_COMPLETE_TRUNCATION_MSG, error=True)
+    return None
+
+
+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", "")
+
+    truncation_err = _check_truncation(file_path, content)
+    if truncation_err is not None:
+        return truncation_err
+
+    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: {os.path.basename(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", "")
+
+    truncation_err = _check_truncation(file_path, content)
+    if truncation_err is not None:
+        return truncation_err
+
+    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/security_hooks_test.py

@@ -63,11 +63,12 @@ def test_read_within_workspace_allowed():
     assert result == {}
 
 
-def test_write_within_workspace_allowed():
+def test_write_builtin_blocked():
+    """SDK built-in Write is blocked — all writes go through MCP Write tool."""
     result = _validate_tool_access(
         "Write", {"file_path": f"{SDK_CWD}/output.json"}, sdk_cwd=SDK_CWD
     )
-    assert result == {}
+    assert _is_denied(result)
 
 
 def test_edit_within_workspace_allowed():