diff --git a/.gitignore b/.gitignore
index 6924a39a..7ae1b28c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -98,11 +98,13 @@ offline-backend-20260512/backend/tests/*
 !offline-backend-20260512/backend/tests/test_deep_research_event_sink.py
 !offline-backend-20260512/backend/tests/test_deep_research_executor_reliability.py
 !offline-backend-20260512/backend/tests/test_deep_research_start_job.py
-!offline-backend-20260512/backend/tests/test_summarization_human_message.py
-!offline-backend-20260512/backend/tests/test_compact_menu_overrides.py
-# Independent enterprise-research persistence, HTTP and race regressions.
-!offline-backend-20260512/backend/tests/test_enterprise_research_tasks.py
-deploy
+!offline-backend-20260512/backend/tests/test_summarization_human_message.py
+!offline-backend-20260512/backend/tests/test_compact_menu_overrides.py
+# Independent enterprise-research persistence, HTTP and race regressions.
+!offline-backend-20260512/backend/tests/test_enterprise_research_tasks.py
+# Position-roundtable per-turn write/present cap (weak-model duplicate artifacts).
+!offline-backend-20260512/backend/tests/test_position_artifact_cap.py
+deploy
 .history
 
 # Cloned third-party repos (local only)
diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md
index af56914d..71376570 100644
--- a/offline-backend-20260512/backend/CLAUDE.md
+++ b/offline-backend-20260512/backend/CLAUDE.md
@@ -534,12 +534,13 @@ that exact answer as the required Markdown artifact in the owning thread's
 **Per-turn outputs write cap (弱模型防重复产物)**: ordinary position-node chats
 and the summary builtin send run-context `position_artifact_cap: 1` (whitelisted
 in `services.merge_run_context_overrides`). `PositionArtifactCapMiddleware`
-then keeps at most one successful `/mnt/user-data/outputs/` `write_file` per
-user turn — truncates excess tool calls in the same model response (and syncs
-raw provider `additional_kwargs.tool_calls`), rejects further writes after an
-`OK` or an earlier sibling in the same batch, ends the turn when only excess
-writes remain, and reminds the model to `present_files` / wrap up. **Action-plan
-does not send the flag** (it must write report `.md` +
+then keeps at most one successful `/mnt/user-data/outputs/` `write_file` **and**
+one `present_files` per user turn — truncates excess tool calls in the same
+model response (and syncs raw provider `additional_kwargs.tool_calls`), rejects
+further writes/presents after an `OK` / `Successfully presented files` or an
+earlier sibling in the same batch, ends the turn when only excess calls remain,
+and reminds the model to present once (or wrap up if already presented).
+**Action-plan does not send the flag** (it must write report `.md` +
 `action-plan-subtasks.json`). Tests: `tests/test_position_artifact_cap.py`.
 
 ### 岗位 (Position) Subscription System
diff --git a/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py b/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py
index 18b9a21c..5c1837fa 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py
@@ -659,8 +659,8 @@ def _build_node_context_text(
         lines.extend(
             [
                 "- 本席位最终只需交付一份 Markdown 报告文档。",
-                "- 请在本轮用 `write_file` **只写一次** `/mnt/user-data/outputs/` 下的单个 `.md` 文件，然后调用 `present_files` 呈现该文件并简短收尾。",
-                "- **禁止**在同一轮再调用 `write_file` 生成第二份（或内容相同的）产物；写完一份后立即结束文件写入。",
+                "- 请在本轮用 `write_file` **只写一次** `/mnt/user-data/outputs/` 下的单个 `.md` 文件，然后调用 `present_files` **只呈现一次**该文件并简短收尾。",
+                "- **禁止**在同一轮再调用 `write_file` 生成第二份（或内容相同的）产物，也**禁止**对同一文件再次 `present_files`。",
                 "- 只有检测到该席位产出的 `.md` 文件后，系统才会把当前席位标记为已完成；仅有文字回答不会完成节点。",
                 "- 报告内容应直接围绕当前岗位职责，吸收任务信息、任务意图与上游席位产物。",
             ]
@@ -1110,7 +1110,7 @@ async def _build_builtin_prompt(
         instructions = [
             "请基于以下岗位会商材料，生成或按用户要求修订《岗位会商方案总结报告》。",
             "必须使用 write_file **只写一次** `/mnt/user-data/outputs/岗位会商方案总结报告.md`，随后在对话中输出完整报告正文。",
-            "**禁止**再次 write_file 生成第二份（或相同）报告；写完后 present_files 并简短收尾即可。",
+            "**禁止**再次 write_file 生成第二份（或相同）报告；写完后 present_files **只调用一次**并简短收尾即可。",
             "报告要覆盖任务目标、岗位关键结论、综合分析、方案建议、风险与执行路径；不要简单拼接原文。",
         ]
     else:
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/position_artifact_cap_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/position_artifact_cap_middleware.py
index 6c861a3f..352a7e1a 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/position_artifact_cap_middleware.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/position_artifact_cap_middleware.py
@@ -1,21 +1,25 @@
-"""Cap ``write_file`` deliveries for position-roundtable seats.
+"""Cap deliverable tools for position-roundtable seats.
 
 Weak models (especially after a first successful Markdown write) often call
-``write_file`` again with another copy of the same report and loop until the
-recursion limit kills the run.
+``write_file`` again with another copy of the same report, or call
+``present_files`` twice for the same path, and loop until the recursion
+limit kills the run. The UI then shows two identical "Successfully
+presented files" cards even though only one file exists on disk.
 
 When the run context carries ``position_artifact_cap`` (set by the position-
 roundtable UI for ordinary seats and the summary agent — **not** for
 action-plan, which must emit report + JSON), this middleware:
 
-1. Truncates excess outputs ``write_file`` tool calls in a single model
-   response (keeps only the first *N*), including raw provider metadata.
-2. Rejects further outputs ``write_file`` calls once *N* successful *or*
-   earlier in-batch writes already exist in the current user turn (so
-   parallel tool calls cannot race past the cap).
-3. Ends the turn when the only remaining calls were excess writes.
-4. Injects a brief hidden reminder before the next model call so the agent
-   wraps up with ``present_files`` / a short reply instead of rewriting.
+1. Truncates excess outputs ``write_file`` **and** ``present_files`` calls in
+   a single model response (keeps only the first *N* of each), including
+   raw provider metadata.
+2. Rejects further such calls once *N* successful *or* earlier in-batch
+   calls already exist in the current user turn (so parallel tool calls
+   cannot race past the cap).
+3. Ends the turn when the only remaining calls were excess writes/presents.
+4. Injects a brief hidden reminder before the next model call:
+   - write done, not yet presented → present once, then wrap up;
+   - write **and** present already done → wrap up, do **not** present again.
 """
 
 from __future__ import annotations
@@ -39,17 +43,29 @@ logger = logging.getLogger(__name__)
 _CONTEXT_KEY = "position_artifact_cap"
 _OUTPUTS_PREFIX = "/mnt/user-data/outputs"
 _WRITE_TOOL = "write_file"
+_PRESENT_TOOL = "present_files"
 
-_REJECT_MSG = (
+_REJECT_WRITE = (
     "本席位本轮产物上限已达：outputs 下只允许写入一份交付文件。"
     "请勿再次调用 write_file；若尚未呈现文件请调用 present_files，"
     "然后用简短文字说明交付已完成。"
 )
 
-_REMINDER = (
+_REJECT_PRESENT = (
+    "本席位本轮产物已经呈现过一次，请勿再次调用 present_files。"
+    "用一两句话说明交付已完成即可，不要再调用任何文件工具。"
+)
+
+_REMINDER_NEED_PRESENT = (
     "【系统提示】本席位本轮产物文件已成功写入 outputs。"
     "禁止再次调用 write_file 生成第二份（或相同）产物。"
-    "请立即 present_files（如尚未呈现），并给出简短收尾回复后结束本轮。"
+    "请立即 present_files（仅此一次，如尚未呈现），并给出简短收尾回复后结束本轮。"
+)
+
+_REMINDER_DONE = (
+    "【系统提示】本席位本轮产物已写入 outputs 并且已经 present_files 呈现过。"
+    "禁止再次调用 write_file 或 present_files。"
+    "请只用一两句话收尾说明交付已完成，不要再调用任何工具。"
 )
 
 
@@ -91,6 +107,17 @@ def _tool_call_writes_outputs(tool_call: Any) -> bool:
     return _is_outputs_path(args.get("path"))
 
 
+def _tool_call_is_present(tool_call: Any) -> bool:
+    return _tool_field(tool_call, "name") == _PRESENT_TOOL
+
+
+def _is_successful_tool_result(content: str) -> bool:
+    text = content.strip()
+    if not text or text.startswith("Error"):
+        return False
+    return True
+
+
 def _cap_from_mapping(mapping: object) -> int | None:
     if not isinstance(mapping, dict):
         return None
@@ -140,31 +167,50 @@ def _current_turn_start(messages: list[Any]) -> int:
     return 0
 
 
-def _successful_outputs_writes(messages: list[Any], *, turn_start: int) -> int:
-    """Count successful outputs ``write_file`` calls in the current user turn."""
-    outputs_ids: set[str] = set()
+def _collect_call_ids(messages: list[Any], *, turn_start: int, match: Callable[[Any], bool]) -> set[str]:
+    ids: set[str] = set()
+    for message in messages[turn_start:]:
+        if getattr(message, "type", None) != "ai":
+            continue
+        for tool_call in getattr(message, "tool_calls", None) or []:
+            if not match(tool_call):
+                continue
+            call_id = _tool_field(tool_call, "id")
+            if isinstance(call_id, str) and call_id:
+                ids.add(call_id)
+    return ids
+
+
+def _count_successful_tools(
+    messages: list[Any],
+    *,
+    turn_start: int,
+    match: Callable[[Any], bool],
+) -> int:
+    matching_ids = _collect_call_ids(messages, turn_start=turn_start, match=match)
     success = 0
     for message in messages[turn_start:]:
-        msg_type = getattr(message, "type", None)
-        if msg_type == "ai":
-            for tool_call in getattr(message, "tool_calls", None) or []:
-                if not _tool_call_writes_outputs(tool_call):
-                    continue
-                call_id = _tool_field(tool_call, "id")
-                if isinstance(call_id, str) and call_id:
-                    outputs_ids.add(call_id)
-            continue
-        if msg_type != "tool":
+        if getattr(message, "type", None) != "tool":
             continue
         call_id = getattr(message, "tool_call_id", None)
-        if not isinstance(call_id, str) or call_id not in outputs_ids:
+        if not isinstance(call_id, str) or call_id not in matching_ids:
             continue
-        content = str(getattr(message, "content", "") or "").strip()
-        if content == "OK" or (content and not content.startswith("Error")):
+        content = str(getattr(message, "content", "") or "")
+        if _is_successful_tool_result(content):
             success += 1
     return success
 
 
+def _successful_outputs_writes(messages: list[Any], *, turn_start: int) -> int:
+    """Count successful outputs ``write_file`` calls in the current user turn."""
+    return _count_successful_tools(messages, turn_start=turn_start, match=_tool_call_writes_outputs)
+
+
+def _successful_presents(messages: list[Any], *, turn_start: int) -> int:
+    """Count successful ``present_files`` calls in the current user turn."""
+    return _count_successful_tools(messages, turn_start=turn_start, match=_tool_call_is_present)
+
+
 def _kept_call_ids(tool_calls: list[Any]) -> set[str]:
     ids: set[str] = set()
     for tool_call in tool_calls:
@@ -203,36 +249,61 @@ def _build_truncated_update(last_msg: Any, kept: list[Any]) -> dict[str, Any]:
     return update
 
 
-def _outputs_budget_used(messages: list[Any], *, turn_start: int, current_id: str | None) -> int:
-    """Successful writes this turn, plus earlier sibling writes in the same AI batch."""
-    used = _successful_outputs_writes(messages, turn_start=turn_start)
+def _in_flight_siblings(
+    messages: list[Any],
+    *,
+    current_id: str | None,
+    match: Callable[[Any], bool],
+) -> int:
+    """Earlier same-batch tool calls that would consume the same budget."""
     if not current_id:
-        return used
+        return 0
     last_ai = None
     for message in reversed(messages):
         if getattr(message, "type", None) == "ai":
             last_ai = message
             break
     if last_ai is None:
-        return used
+        return 0
     batch = list(getattr(last_ai, "tool_calls", None) or [])
     if current_id not in {str(_tool_field(tool_call, "id") or "") for tool_call in batch}:
-        return used
+        return 0
+    used = 0
     for tool_call in batch:
         call_id = str(_tool_field(tool_call, "id") or "")
         if call_id == current_id:
             break
-        if _tool_call_writes_outputs(tool_call):
+        if match(tool_call):
             used += 1
     return used
 
 
+def _outputs_budget_used(messages: list[Any], *, turn_start: int, current_id: str | None) -> int:
+    return _successful_outputs_writes(messages, turn_start=turn_start) + _in_flight_siblings(
+        messages, current_id=current_id, match=_tool_call_writes_outputs
+    )
+
+
+def _present_budget_used(messages: list[Any], *, turn_start: int, current_id: str | None) -> int:
+    return _successful_presents(messages, turn_start=turn_start) + _in_flight_siblings(
+        messages, current_id=current_id, match=_tool_call_is_present
+    )
+
+
+def _reminder_for(*, writes_done: bool, presents_done: bool) -> str | None:
+    if not writes_done:
+        return None
+    if presents_done:
+        return _REMINDER_DONE
+    return _REMINDER_NEED_PRESENT
+
+
 class PositionArtifactCapMiddleware(AgentMiddleware[AgentState]):
-    """Enforce a per-turn outputs ``write_file`` budget for position roundtable."""
+    """Enforce a per-turn outputs write/present budget for position roundtable."""
 
     state_schema = AgentState
 
-    def _truncate_excess_writes(self, state: AgentState, runtime: Runtime) -> dict | None:
+    def _truncate_excess(self, state: AgentState, runtime: Runtime) -> dict | None:
         cap = _cap_from_runtime(runtime)
         if cap is None:
             return None
@@ -249,43 +320,52 @@ class PositionArtifactCapMiddleware(AgentMiddleware[AgentState]):
         if not tool_calls:
             return None
 
-        already = _successful_outputs_writes(messages[:-1], turn_start=_current_turn_start(messages))
-        remaining = max(0, cap - already)
+        turn_start = _current_turn_start(messages)
+        prior = messages[:-1]
+        remaining_writes = max(0, cap - _successful_outputs_writes(prior, turn_start=turn_start))
+        remaining_presents = max(0, cap - _successful_presents(prior, turn_start=turn_start))
 
         kept: list[Any] = []
         dropped = 0
-        kept_outputs = 0
+        kept_writes = 0
+        kept_presents = 0
         for tool_call in tool_calls:
             if _tool_call_writes_outputs(tool_call):
-                if kept_outputs >= remaining:
+                if kept_writes >= remaining_writes:
                     dropped += 1
                     continue
-                kept_outputs += 1
+                kept_writes += 1
+            elif _tool_call_is_present(tool_call):
+                if kept_presents >= remaining_presents:
+                    dropped += 1
+                    continue
+                kept_presents += 1
             kept.append(tool_call)
 
         if dropped == 0:
             return None
 
         logger.warning(
-            "Position artifact cap truncated %s excess write_file call(s) (cap=%s, already=%s)",
+            "Position artifact cap truncated %s excess write_file/present_files call(s) "
+            "(cap=%s, remaining_writes=%s, remaining_presents=%s)",
             dropped,
             cap,
-            already,
+            remaining_writes,
+            remaining_presents,
         )
         updated = last_msg.model_copy(update=_build_truncated_update(last_msg, kept))
         result: dict[str, Any] = {"messages": [updated]}
         if not kept:
             # No remaining tools — stop instead of looping back to the model
-            # which would just emit another copy of write_file.
+            # which would just emit another copy of write_file / present_files.
             result["jump_to"] = "end"
         return result
 
-    def _reject_over_cap(
-        self,
-        request: ToolCallRequest,
-    ) -> ToolMessage | None:
+    def _reject_over_cap(self, request: ToolCallRequest) -> ToolMessage | None:
         tool_call = getattr(request, "tool_call", None) or {}
-        if not _tool_call_writes_outputs(tool_call):
+        is_write = _tool_call_writes_outputs(tool_call)
+        is_present = _tool_call_is_present(tool_call)
+        if not is_write and not is_present:
             return None
 
         runtime = getattr(request, "runtime", None)
@@ -296,20 +376,29 @@ class PositionArtifactCapMiddleware(AgentMiddleware[AgentState]):
         state = getattr(request, "state", None) or {}
         messages = list(state.get("messages") or []) if isinstance(state, dict) else []
         current_id = str(_tool_field(tool_call, "id") or "")
-        used = _outputs_budget_used(messages, turn_start=_current_turn_start(messages), current_id=current_id or None)
+        turn_start = _current_turn_start(messages)
+        if is_write:
+            used = _outputs_budget_used(messages, turn_start=turn_start, current_id=current_id or None)
+            reject_text = _REJECT_WRITE
+            tool_name = _WRITE_TOOL
+        else:
+            used = _present_budget_used(messages, turn_start=turn_start, current_id=current_id or None)
+            reject_text = _REJECT_PRESENT
+            tool_name = _PRESENT_TOOL
         if used < cap:
             return None
 
         logger.warning(
-            "Position artifact cap rejected write_file (cap=%s, used=%s, id=%s)",
+            "Position artifact cap rejected %s (cap=%s, used=%s, id=%s)",
+            tool_name,
             cap,
             used,
             current_id,
         )
         return ToolMessage(
-            content=f"Error: {_REJECT_MSG}",
+            content=f"Error: {reject_text}",
             tool_call_id=current_id,
-            name=_WRITE_TOOL,
+            name=tool_name,
         )
 
     def _maybe_remind(self, request: ModelRequest) -> ModelRequest:
@@ -317,11 +406,14 @@ class PositionArtifactCapMiddleware(AgentMiddleware[AgentState]):
         if cap is None:
             return request
         messages = list(getattr(request, "messages", None) or [])
-        already = _successful_outputs_writes(messages, turn_start=_current_turn_start(messages))
-        if already < cap:
+        turn_start = _current_turn_start(messages)
+        writes_done = _successful_outputs_writes(messages, turn_start=turn_start) >= cap
+        presents_done = _successful_presents(messages, turn_start=turn_start) >= cap
+        text = _reminder_for(writes_done=writes_done, presents_done=presents_done)
+        if text is None:
             return request
         reminder = HumanMessage(
-            content=_REMINDER,
+            content=text,
             name="position_artifact_cap",
             additional_kwargs={"hide_from_ui": True, "lc_source": "position_artifact_cap"},
         )
@@ -330,12 +422,12 @@ class PositionArtifactCapMiddleware(AgentMiddleware[AgentState]):
     @hook_config(can_jump_to=["end"])
     @override
     def after_model(self, state: AgentState, runtime: Runtime) -> dict | None:
-        return self._truncate_excess_writes(state, runtime)
+        return self._truncate_excess(state, runtime)
 
     @hook_config(can_jump_to=["end"])
     @override
     async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict | None:
-        return self._truncate_excess_writes(state, runtime)
+        return self._truncate_excess(state, runtime)
 
     @override
     def wrap_tool_call(
diff --git a/offline-backend-20260512/backend/tests/test_position_artifact_cap.py b/offline-backend-20260512/backend/tests/test_position_artifact_cap.py
new file mode 100644
index 00000000..39bf65cd
--- /dev/null
+++ b/offline-backend-20260512/backend/tests/test_position_artifact_cap.py
@@ -0,0 +1,273 @@
+"""Tests for PositionArtifactCapMiddleware."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+
+from deerflow.agents.middlewares.position_artifact_cap_middleware import (
+    PositionArtifactCapMiddleware,
+    _reminder_for,
+    _successful_outputs_writes,
+    _successful_presents,
+)
+
+
+def _runtime(cap: int | None = 1) -> SimpleNamespace:
+    context: dict = {"thread_id": "t-pos"}
+    if cap is not None:
+        context["position_artifact_cap"] = cap
+    return SimpleNamespace(context=context, config={"configurable": dict(context)})
+
+
+def _write_call(call_id: str, path: str = "/mnt/user-data/outputs/report.md") -> dict:
+    return {
+        "id": call_id,
+        "name": "write_file",
+        "args": {"path": path, "content": "# hello", "description": "deliver"},
+    }
+
+
+def _present_call(call_id: str, path: str = "/mnt/user-data/outputs/report.md") -> dict:
+    return {
+        "id": call_id,
+        "name": "present_files",
+        "args": {"filepaths": [path]},
+    }
+
+
+def test_successful_outputs_writes_counts_ok_in_current_turn_only() -> None:
+    messages = [
+        HumanMessage(content="上一轮"),
+        AIMessage(content="", tool_calls=[_write_call("old")]),
+        ToolMessage(content="OK", tool_call_id="old", name="write_file"),
+        HumanMessage(content="本轮请交付"),
+        AIMessage(content="", tool_calls=[_write_call("new")]),
+        ToolMessage(content="OK", tool_call_id="new", name="write_file"),
+    ]
+    assert _successful_outputs_writes(messages, turn_start=3) == 1
+
+
+def test_after_model_truncates_second_outputs_write_in_same_response() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付一份报告"),
+            AIMessage(
+                content="正在写入",
+                tool_calls=[
+                    _write_call("w1"),
+                    _write_call("w2", "/mnt/user-data/outputs/report-copy.md"),
+                    _present_call("p1"),
+                ],
+            ),
+        ]
+    }
+    update = mw.after_model(state, _runtime(1))
+    assert update is not None
+    tool_calls = update["messages"][0].tool_calls
+    assert [tc["id"] for tc in tool_calls] == ["w1", "p1"]
+
+
+def test_after_model_strips_all_writes_once_cap_already_met() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付"),
+            AIMessage(content="", tool_calls=[_write_call("done")]),
+            ToolMessage(content="OK", tool_call_id="done", name="write_file"),
+            AIMessage(content="再写一份", tool_calls=[_write_call("again")]),
+        ]
+    }
+    update = mw.after_model(state, _runtime(1))
+    assert update is not None
+    assert update["messages"][0].tool_calls == []
+    assert update.get("jump_to") == "end"
+    assert "产物上限" not in str(update["messages"][0].content)
+    assert "tool_calls" not in (update["messages"][0].additional_kwargs or {})
+
+
+def test_after_model_noop_without_cap() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付"),
+            AIMessage(content="", tool_calls=[_write_call("w1"), _write_call("w2")]),
+        ]
+    }
+    assert mw.after_model(state, _runtime(None)) is None
+
+
+def test_wrap_tool_call_rejects_after_successful_delivery() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付"),
+            AIMessage(content="", tool_calls=[_write_call("done")]),
+            ToolMessage(content="OK", tool_call_id="done", name="write_file"),
+        ]
+    }
+    request = SimpleNamespace(
+        tool_call=_write_call("again"),
+        runtime=_runtime(1),
+        state=state,
+    )
+    called = {"n": 0}
+
+    def handler(_request):
+        called["n"] += 1
+        return ToolMessage(content="OK", tool_call_id="again", name="write_file")
+
+    result = mw.wrap_tool_call(request, handler)
+    assert called["n"] == 0
+    assert isinstance(result, ToolMessage)
+    assert str(result.content).startswith("Error:")
+
+
+def test_wrap_tool_call_allows_first_write() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {"messages": [HumanMessage(content="请交付")]}
+    request = SimpleNamespace(
+        tool_call=_write_call("w1"),
+        runtime=_runtime(1),
+        state=state,
+    )
+
+    def handler(_request):
+        return ToolMessage(content="OK", tool_call_id="w1", name="write_file")
+
+    result = mw.wrap_tool_call(request, handler)
+    assert result.content == "OK"
+
+
+def test_wrap_tool_call_rejects_parallel_second_write_in_same_batch() -> None:
+    """Two write_file in one AIMessage can run concurrently; order in the list is the budget."""
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付"),
+            AIMessage(content="", tool_calls=[_write_call("w1"), _write_call("w2")]),
+        ]
+    }
+    first = SimpleNamespace(tool_call=_write_call("w1"), runtime=_runtime(1), state=state)
+    second = SimpleNamespace(tool_call=_write_call("w2"), runtime=_runtime(1), state=state)
+
+    def handler(request):
+        return ToolMessage(content="OK", tool_call_id=request.tool_call["id"], name="write_file")
+
+    assert mw.wrap_tool_call(first, handler).content == "OK"
+    rejected = mw.wrap_tool_call(second, handler)
+    assert str(rejected.content).startswith("Error:")
+
+
+def test_after_model_clears_provider_tool_call_metadata() -> None:
+    mw = PositionArtifactCapMiddleware()
+    extra = {
+        "tool_calls": [
+            {"id": "w1", "type": "function", "function": {"name": "write_file"}},
+            {"id": "w2", "type": "function", "function": {"name": "write_file"}},
+        ]
+    }
+    ai = AIMessage(content="写两份", tool_calls=[_write_call("w1"), _write_call("w2")])
+    ai.additional_kwargs = extra
+    state = {"messages": [HumanMessage(content="请交付"), ai]}
+    update = mw.after_model(state, _runtime(1))
+    assert update is not None
+    kept = update["messages"][0]
+    assert [tc["id"] for tc in kept.tool_calls] == ["w1"]
+    assert [item["id"] for item in kept.additional_kwargs.get("tool_calls", [])] == ["w1"]
+
+
+def test_action_plan_style_cap_two_allows_second_write() -> None:
+    """Action-plan is exempt in the UI (no flag); if a cap of 2 were set it must allow both."""
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="生成行动规划"),
+            AIMessage(content="", tool_calls=[_write_call("md", "/mnt/user-data/outputs/行动规划报告.md")]),
+            ToolMessage(content="OK", tool_call_id="md", name="write_file"),
+            AIMessage(
+                content="",
+                tool_calls=[_write_call("json", "/mnt/user-data/outputs/action-plan-subtasks.json")],
+            ),
+        ]
+    }
+    assert mw.after_model(state, _runtime(2)) is None
+
+
+def test_after_model_truncates_second_present_files_in_same_response() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付"),
+            AIMessage(
+                content="",
+                tool_calls=[_write_call("w1"), _present_call("p1"), _present_call("p2")],
+            ),
+        ]
+    }
+    update = mw.after_model(state, _runtime(1))
+    assert update is not None
+    assert [tc["id"] for tc in update["messages"][0].tool_calls] == ["w1", "p1"]
+
+
+def test_after_model_strips_second_present_after_first_succeeded() -> None:
+    """Screenshot bug: write+present already done, weak model presents the same file again."""
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请评估"),
+            AIMessage(content="", tool_calls=[_write_call("w1"), _present_call("p1")]),
+            ToolMessage(content="OK", tool_call_id="w1", name="write_file"),
+            ToolMessage(content="Successfully presented files", tool_call_id="p1", name="present_files"),
+            AIMessage(content="评估报告已交付", tool_calls=[_present_call("p2")]),
+        ]
+    }
+    update = mw.after_model(state, _runtime(1))
+    assert update is not None
+    assert update["messages"][0].tool_calls == []
+    assert update.get("jump_to") == "end"
+
+
+def test_wrap_tool_call_rejects_second_present_files() -> None:
+    mw = PositionArtifactCapMiddleware()
+    state = {
+        "messages": [
+            HumanMessage(content="请交付"),
+            AIMessage(content="", tool_calls=[_present_call("p1")]),
+            ToolMessage(content="Successfully presented files", tool_call_id="p1", name="present_files"),
+        ]
+    }
+    request = SimpleNamespace(
+        tool_call=_present_call("p2"),
+        runtime=_runtime(1),
+        state=state,
+    )
+    called = {"n": 0}
+
+    def handler(_request):
+        called["n"] += 1
+        return ToolMessage(content="Successfully presented files", tool_call_id="p2", name="present_files")
+
+    result = mw.wrap_tool_call(request, handler)
+    assert called["n"] == 0
+    assert isinstance(result, ToolMessage)
+    assert "已经呈现" in str(result.content)
+
+
+def test_reminder_asks_present_only_when_not_yet_presented() -> None:
+    assert "present_files" in (_reminder_for(writes_done=True, presents_done=False) or "")
+    done = _reminder_for(writes_done=True, presents_done=True) or ""
+    assert "已经 present_files" in done
+    assert "请立即 present_files" not in done
+    assert _reminder_for(writes_done=False, presents_done=False) is None
+
+
+def test_successful_presents_counts_tool_result() -> None:
+    messages = [
+        HumanMessage(content="请交付"),
+        AIMessage(content="", tool_calls=[_present_call("p1")]),
+        ToolMessage(content="Successfully presented files", tool_call_id="p1", name="present_files"),
+    ]
+    assert _successful_presents(messages, turn_start=0) == 1