diff --git a/backend/app/services/agent_tools.py b/backend/app/services/agent_tools.py
index b4665f33c..700ae4ca0 100644
--- a/backend/app/services/agent_tools.py
+++ b/backend/app/services/agent_tools.py
@@ -80,6 +80,7 @@
 from app.services.builtin_tool_definitions import (
     BUILTIN_TOOL_DEFINITIONS,
     BUILTIN_TOOL_NAMES,
+    WRITE_FILE_MAX_CONTENT_CHARS,
     builtin_model_definition,
     builtin_model_definitions,
     builtin_readiness,
@@ -1690,10 +1691,20 @@ async def _execute_workspace_mutation(
     if tool_name == "write_file":
         path = arguments.get("path")
         content = arguments.get("content")
+        mode = arguments.get("mode", "overwrite")
         if not path:
             return "❌ Missing required argument 'path' for write_file. Please provide a file path like 'skills/my-skill/SKILL.md'"
         if content is None:
             return "❌ Missing required argument 'content' for write_file"
+        if not isinstance(content, str):
+            return "❌ write_file content must be a string"
+        if mode not in {"overwrite", "append"}:
+            return "❌ write_file mode must be overwrite or append"
+        if len(content) > WRITE_FILE_MAX_CONTENT_CHARS:
+            return (
+                "❌ write_file content exceeds 6000 characters. Write the first "
+                "chunk with mode=overwrite, then append one smaller chunk per later turn."
+            )
         if is_focus_file_path(path):
             return "❌ Focus is no longer stored in focus.md. Use upsert_focus_item or complete_focus_item."
         if _is_enterprise_info_path(path):
@@ -1710,13 +1721,10 @@ async def _execute_workspace_mutation(
                 operation="write",
                 session_id=session_id,
                 enforce_human_lock=True,
+                append=mode == "append",
             )
             await _wdb.commit()
-        return (
-            f"✅ Written to {write_result.path} ({len(content)} chars)"
-            if write_result.ok
-            else f"❌ {write_result.message}"
-        )
+        return f"✅ {write_result.message}" if write_result.ok else f"❌ {write_result.message}"
 
     if tool_name == "move_file":
         source_path = arguments.get("source_path")
@@ -2103,6 +2111,7 @@ async def _write_file_outcome(
     """Write one workspace file using the structured collaboration result."""
     path = arguments.get("path")
     content = arguments.get("content")
+    mode = arguments.get("mode", "overwrite")
     if not isinstance(path, str) or not path.strip() or content is None:
         return _typed_failure(
             "write_file requires non-empty path and content.",
@@ -2113,6 +2122,17 @@ async def _write_file_outcome(
             "write_file content must be a string.",
             "invalid_tool_arguments",
         )
+    if mode not in {"overwrite", "append"}:
+        return _typed_failure(
+            "write_file mode must be overwrite or append.",
+            "invalid_tool_arguments",
+        )
+    if len(content) > WRITE_FILE_MAX_CONTENT_CHARS:
+        return _typed_failure(
+            "write_file content exceeds 6000 characters. Write the first chunk "
+            "with mode=overwrite, then append one smaller chunk per later turn.",
+            "write_file_content_too_large",
+        )
     if is_focus_file_path(path):
         return _typed_failure(
             "Focus is structured data; use upsert_focus_item.",
@@ -2138,6 +2158,7 @@ async def _write_file_outcome(
                 operation="write",
                 session_id=session_id,
                 enforce_human_lock=True,
+                append=mode == "append",
             )
             if not write_result.ok:
                 return _typed_failure(
@@ -2155,9 +2176,7 @@ async def _write_file_outcome(
             f"Workspace write failed: {type(exc).__name__}",
             "workspace_write_failed",
         )
-    return _typed_success(
-        f"Written to {write_result.path} ({len(content)} chars)."
-    )
+    return _typed_success(f"{write_result.message}.")
 
 
 async def _list_files_outcome(
diff --git a/backend/app/services/builtin_tool_definitions.py b/backend/app/services/builtin_tool_definitions.py
index 2ae4cff1f..6700d4ba2 100644
--- a/backend/app/services/builtin_tool_definitions.py
+++ b/backend/app/services/builtin_tool_definitions.py
@@ -16,6 +16,9 @@
 from app.services.llm.finish import FINISH_TOOL_SEED
 
 
+WRITE_FILE_MAX_CONTENT_CHARS = 6_000
+
+
 # Builtin tool definitions — these map to the hardcoded AGENT_TOOLS
 _BUILTIN_TOOL_SOURCE = [
     FINISH_TOOL_SEED,
@@ -119,7 +122,7 @@
     {
         "name": "write_file",
         "display_name": "Write File",
-        "description": "Write or update a file in the workspace. Before creating a new document under workspace/, first inspect the relevant directories with list_files, prefer an existing topical subfolder over the workspace root, and create a new subfolder when the content belongs to a new category. Avoid placing standalone document files directly in workspace/ root unless the user explicitly wants that. Can update memory/memory.md, create documents in workspace/, create skills in skills/.",
+        "description": "Write or incrementally append UTF-8 text to a file in the workspace. Each call accepts at most 6000 content characters. For a longer generated file such as HTML, CSS, JavaScript, or markdown, call write_file once with mode=overwrite for the first chunk, then use one mode=append call per later model turn for each remaining chunk; never emit the whole file or multiple large chunks in one response. Before creating a new document under workspace/, first inspect the relevant directories with list_files, prefer an existing topical subfolder over the workspace root, and create a new subfolder when the content belongs to a new category. Avoid placing standalone document files directly in workspace/ root unless the user explicitly wants that. Can update memory/memory.md, create documents in workspace/, create skills in skills/.",
         "category": "file",
         "icon": "✏️",
         "is_default": True,
@@ -127,7 +130,17 @@
             "type": "object",
             "properties": {
                 "path": {"type": "string", "description": "File path, e.g.: memory/memory.md, workspace/reports/report.md, workspace/knowledge_base/notes.md. Prefer a meaningful subfolder instead of writing loose files into workspace/ root."},
-                "content": {"type": "string", "description": "File content to write"},
+                "content": {
+                    "type": "string",
+                    "maxLength": WRITE_FILE_MAX_CONTENT_CHARS,
+                    "description": "One file-content chunk, at most 6000 characters. Keep long generated content split across later tool turns.",
+                },
+                "mode": {
+                    "type": "string",
+                    "enum": ["overwrite", "append"],
+                    "default": "overwrite",
+                    "description": "overwrite creates or replaces the file (default); append adds this chunk to an existing file after the previous write succeeds.",
+                },
             },
             "required": ["path", "content"],
         },
@@ -4040,6 +4053,7 @@ def validate_builtin_tool_definitions() -> None:
     "BUILTIN_TOOL_SEEDS",
     "GROUP_BUILTIN_TOOL_DEFINITIONS",
     "GROUP_RUNTIME_TOOL_DEFINITIONS",
+    "WRITE_FILE_MAX_CONTENT_CHARS",
     "builtin_model_definition",
     "builtin_model_definitions",
     "builtin_cross_space_action",
diff --git a/backend/app/services/llm/caller.py b/backend/app/services/llm/caller.py
index 742aadf0d..3f090a1d2 100644
--- a/backend/app/services/llm/caller.py
+++ b/backend/app/services/llm/caller.py
@@ -65,10 +65,12 @@ async def execute_tool(*args, **kwargs):
 WRITE_FILE_PROTOCOL_REPAIR_COUNTER_KEY = "invalid_tool_call:write_file"
 WRITE_FILE_PROTOCOL_REPAIR_INSTRUCTION = (
     "Your previous `write_file` call was not executed because `function.arguments` "
-    "was invalid JSON or was truncated. Retry `write_file` with "
-    "`function.arguments` as one valid JSON object string. Do not repeat the same "
-    "oversized whole-file content; reduce the content in this call and continue with "
-    "later normal tool calls if needed. Do not explain; only retry with a valid tool call."
+    "was invalid JSON or was truncated. Do not retry the entire file. Retry now with "
+    "one valid JSON object containing only the first content chunk, at most 6000 "
+    "characters, and set mode=overwrite. After that tool call succeeds, continue in "
+    "later turns with exactly one smaller chunk per call using mode=append. Escape "
+    "quotes and newlines in each chunk. Do not explain; only issue the first smaller "
+    "tool call."
 )
 WRITE_FILE_PROTOCOL_FAILURE_MESSAGE = (
     "本次文件生成未完成：write_file 工具参数无效或被截断，连续重试后仍无法执行。"
diff --git a/backend/app/services/workspace_collaboration.py b/backend/app/services/workspace_collaboration.py
index d0dec57e4..4ac7d449e 100644
--- a/backend/app/services/workspace_collaboration.py
+++ b/backend/app/services/workspace_collaboration.py
@@ -547,8 +547,9 @@ async def write_workspace_file(
     enforce_human_lock: bool = True,
     merge_user_autosave: bool = False,
     expected_version_token: str | None = None,
+    append: bool = False,
 ) -> WorkspaceWriteResult:
-    """Write text content, enforcing human locks for agent/system actors."""
+    """Write or append text content, enforcing human locks for agent/system actors."""
     normalized = normalize_workspace_path(path)
     if not normalized:
         return WorkspaceWriteResult(False, normalized, "Missing file path")
@@ -568,17 +569,34 @@ async def write_workspace_file(
 
     storage = get_storage_backend()
     storage_key = normalize_storage_key(f"{agent_id}/{normalized}")
+    current_version = await storage.get_version(storage_key)
+    if append and not current_version.exists:
+        return WorkspaceWriteResult(
+            False,
+            normalized,
+            f"Cannot append to missing file: {normalized}",
+        )
     local_base_available = _should_mirror_to_local_filesystem(storage)
     try:
         target = safe_agent_path(base_dir, normalized)
     except Exception:
         target = None
         local_base_available = False
-    before = await storage.read_text(storage_key, encoding="utf-8", errors="replace") if await storage.exists(storage_key) else None
+    before = (
+        await storage.read_text(storage_key, encoding="utf-8", errors="replace")
+        if current_version.exists
+        else None
+    )
+    after = f"{before or ''}{content}" if append else content
+    condition = None
+    if expected_version_token is not None:
+        condition = WriteCondition(version_token=expected_version_token)
+    elif append:
+        condition = WriteCondition(version_token=current_version.token)
     write_result = await storage.write_bytes_if_match(
         storage_key,
-        content.encode("utf-8"),
-        condition=WriteCondition(version_token=expected_version_token) if expected_version_token is not None else None,
+        after.encode("utf-8"),
+        condition=condition,
         content_type="text/plain; charset=utf-8",
     )
     if not write_result.ok:
@@ -586,7 +604,7 @@ async def write_workspace_file(
     if local_base_available and target is not None:
         target.parent.mkdir(parents=True, exist_ok=True)
         async with aiofiles.open(target, "w", encoding="utf-8") as f:
-            await f.write(content)
+            await f.write(after)
 
     revision = await record_revision(
         db,
@@ -596,14 +614,18 @@ async def write_workspace_file(
         actor_type=actor_type,
         actor_id=actor_id,
         before_content=before,
-        after_content=content,
+        after_content=after,
         session_id=session_id,
         merge_user_autosave=merge_user_autosave,
     )
     return WorkspaceWriteResult(
         True,
         normalized,
-        f"Written to {normalized} ({len(content)} chars)",
+        (
+            f"Appended to {normalized} ({len(content)} chars; {len(after)} total)"
+            if append
+            else f"Written to {normalized} ({len(content)} chars)"
+        ),
         revision_id=str(revision.id) if revision else None,
     )
 
diff --git a/backend/tests/test_agent_tools_storage_workspace.py b/backend/tests/test_agent_tools_storage_workspace.py
index 36d345ddc..3d27566aa 100644
--- a/backend/tests/test_agent_tools_storage_workspace.py
+++ b/backend/tests/test_agent_tools_storage_workspace.py
@@ -212,6 +212,96 @@ async def _noop_revision(*args, **kwargs):
     assert not (tmp_path / str(agent_id) / "workspace" / "test.md").exists()
 
 
+@pytest.mark.asyncio
+async def test_write_workspace_file_appends_with_version_guard(monkeypatch, tmp_path):
+    agent_id = uuid.uuid4()
+    storage = MemoryStorageBackend({
+        f"{agent_id}/workspace/page.html": b"<main>",
+    })
+    monkeypatch.setattr(workspace_collaboration, "get_storage_backend", lambda: storage)
+    revisions = []
+
+    async def _record_revision(*args, **kwargs):
+        revisions.append(kwargs)
+        return None
+
+    monkeypatch.setattr(workspace_collaboration, "record_revision", _record_revision)
+
+    result = await workspace_collaboration.write_workspace_file(
+        db=None,
+        agent_id=agent_id,
+        base_dir=tmp_path / str(agent_id),
+        path="workspace/page.html",
+        content="content</main>",
+        actor_type="agent",
+        actor_id=agent_id,
+        enforce_human_lock=False,
+        append=True,
+    )
+
+    assert result.ok is True
+    assert result.message == "Appended to workspace/page.html (14 chars; 20 total)"
+    assert storage.files[f"{agent_id}/workspace/page.html"] == b"<main>content</main>"
+    assert revisions[0]["before_content"] == "<main>"
+    assert revisions[0]["after_content"] == "<main>content</main>"
+
+
+@pytest.mark.asyncio
+async def test_write_workspace_file_rejects_append_to_missing_file(monkeypatch, tmp_path):
+    agent_id = uuid.uuid4()
+    storage = MemoryStorageBackend()
+    monkeypatch.setattr(workspace_collaboration, "get_storage_backend", lambda: storage)
+
+    result = await workspace_collaboration.write_workspace_file(
+        db=None,
+        agent_id=agent_id,
+        base_dir=tmp_path / str(agent_id),
+        path="workspace/page.html",
+        content="content",
+        actor_type="agent",
+        actor_id=agent_id,
+        enforce_human_lock=False,
+        append=True,
+    )
+
+    assert result.ok is False
+    assert result.message == "Cannot append to missing file: workspace/page.html"
+    assert storage.files == {}
+
+
+@pytest.mark.asyncio
+async def test_write_workspace_file_append_does_not_overwrite_a_concurrent_change(
+    monkeypatch,
+    tmp_path,
+):
+    agent_id = uuid.uuid4()
+    key = f"{agent_id}/workspace/page.html"
+
+    class RacingStorageBackend(MemoryStorageBackend):
+        async def write_bytes_if_match(self, storage_key, data, **kwargs):
+            await self.write_bytes(storage_key, b"concurrent")
+            return await super().write_bytes_if_match(storage_key, data, **kwargs)
+
+    storage = RacingStorageBackend({key: b"first"})
+    monkeypatch.setattr(workspace_collaboration, "get_storage_backend", lambda: storage)
+
+    result = await workspace_collaboration.write_workspace_file(
+        db=None,
+        agent_id=agent_id,
+        base_dir=tmp_path / str(agent_id),
+        path="workspace/page.html",
+        content=" second",
+        actor_type="agent",
+        actor_id=agent_id,
+        enforce_human_lock=False,
+        append=True,
+    )
+
+    assert result.ok is False
+    assert result.message == "Conflict detected while writing workspace/page.html"
+    assert storage.files[key] == b"concurrent"
+
+
 @pytest.mark.asyncio
 async def test_flush_temp_workspace_only_writes_changed_files(monkeypatch):
     agent_id = uuid.uuid4()
diff --git a/backend/tests/test_builtin_tool_contracts.py b/backend/tests/test_builtin_tool_contracts.py
index 82adb6b7b..1bb938238 100644
--- a/backend/tests/test_builtin_tool_contracts.py
+++ b/backend/tests/test_builtin_tool_contracts.py
@@ -87,6 +87,7 @@ def test_builtin_model_definition_ignores_stale_database_contract() -> None:
 
 
 def test_known_schema_contracts_match_handler_validation() -> None:
+    write_file = builtin_model_definition("write_file")["function"]["parameters"]
     send_channel = builtin_model_definition("send_channel_message")["function"]["parameters"]
     send_platform = builtin_model_definition("send_platform_message")["function"]["parameters"]
     upload_image = builtin_model_definition("upload_image")["function"]["parameters"]
@@ -94,6 +95,10 @@ def test_known_schema_contracts_match_handler_validation() -> None:
     set_trigger = builtin_model_definition("set_trigger")["function"]["parameters"]
     import_mcp = builtin_model_definition("import_mcp_server")["function"]["parameters"]
 
+    assert write_file["properties"]["content"]["maxLength"] == 6_000
+    assert write_file["properties"]["mode"]["enum"] == ["overwrite", "append"]
+    assert write_file["properties"]["mode"]["default"] == "overwrite"
+    assert write_file["required"] == ["path", "content"]
     assert send_channel["required"] == ["target_member_id", "message"]
     assert send_platform["required"] == ["message"]
     assert send_platform["anyOf"] == [
@@ -305,6 +310,82 @@ async def test_typed_builtin_validation_failure_is_explicit_not_unknown() -> Non
     assert outcome.error_code == "invalid_tool_arguments"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("arguments", "error_code"),
+    [
+        (
+            {"path": "workspace/page.html", "content": "x" * 6_001},
+            "write_file_content_too_large",
+        ),
+        (
+            {
+                "path": "workspace/page.html",
+                "content": "chunk",
+                "mode": "replace",
+            },
+            "invalid_tool_arguments",
+        ),
+    ],
+)
+async def test_write_file_enforces_incremental_write_contract(
+    arguments: dict,
+    error_code: str,
+) -> None:
+    outcome = await agent_tools.execute_builtin_tool_outcome(
+        "write_file",
+        arguments,
+        agent_id=uuid.uuid4(),
+        user_id=uuid.uuid4(),
+    )
+
+    assert outcome.status == "failed"
+    assert outcome.error_code == error_code
+    assert "append" in (outcome.result_summary or "")
+
+
+@pytest.mark.asyncio
+async def test_write_file_append_mode_reaches_the_workspace_boundary(monkeypatch, tmp_path) -> None:
+    agent_id = uuid.uuid4()
+    recorded = {}
+
+    class _WriteSession:
+        async def commit(self) -> None:
+            recorded["committed"] = True
+
+    @asynccontextmanager
+    async def _session_factory():
+        yield _WriteSession()
+
+    async def _write_workspace_file(db, **kwargs):
+        recorded.update(kwargs)
+        return SimpleNamespace(
+            ok=True,
+            path=kwargs["path"],
+            message="Appended to workspace/page.html (5 chars; 10 total)",
+        )
+
+    monkeypatch.setattr(agent_tools, "async_session", _session_factory)
+    monkeypatch.setattr(agent_tools, "write_workspace_file", _write_workspace_file)
+
+    outcome = await agent_tools._write_file_outcome(
+        agent_id,
+        {
+            "path": "workspace/page.html",
+            "content": "later",
+            "mode": "append",
+        },
+        base_dir=tmp_path,
+        session_id="session-1",
+    )
+
+    assert outcome.status == "succeeded"
+    assert recorded["append"] is True
+    assert recorded["operation"] == "write"
+    assert recorded["session_id"] == "session-1"
+    assert recorded["committed"] is True
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("tool_name", "arguments"),
diff --git a/backend/tests/test_llm_single_step.py b/backend/tests/test_llm_single_step.py
index 931bd51eb..fd9b628b1 100644
--- a/backend/tests/test_llm_single_step.py
+++ b/backend/tests/test_llm_single_step.py
@@ -212,7 +212,10 @@ async def test_complete_once_returns_a_bounded_repair_instruction_for_invalid_ar
     assert result.retry_instruction is not None
     assert "valid JSON" in result.retry_instruction
     assert "not executed" in result.retry_instruction
-    assert "same oversized whole-file content" in result.retry_instruction
+    assert "Do not retry the entire file" in result.retry_instruction
+    assert "6000 characters" in result.retry_instruction
+    assert "mode=overwrite" in result.retry_instruction
+    assert "mode=append" in result.retry_instruction
     assert result.retry_tool_name == "write_file"
     assert client.closed is True