diff --git a/backend/packages/harness/deerflow/subagents/context_snapshot.py b/backend/packages/harness/deerflow/subagents/context_snapshot.py
index e68df0f67a..4456529aef 100644
--- a/backend/packages/harness/deerflow/subagents/context_snapshot.py
+++ b/backend/packages/harness/deerflow/subagents/context_snapshot.py
@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import logging
+from collections import Counter
 from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any
@@ -118,11 +119,14 @@ class ParentContextSnapshot:
         # Providers may reuse call ids across turns. Match each result to the
         # preceding call, including hidden frames so their results cannot be
         # reassigned to a visible call. Only retained pairs count as completed.
-        call_positions: dict[str, int] = {}
+        call_positions: dict[str, int | None] = {}
         completed_calls: set[tuple[int, str]] = set()
         for index, message in enumerate(messages):
             if isinstance(message, AIMessage):
-                call_positions.update((call["id"], index) for call in message.tool_calls)
+                # A result cannot identify one of several calls with the same
+                # id in this message. Claim the id without completing any call,
+                # so it cannot fall back to an earlier occurrence either.
+                call_positions.update((call_id, index if count == 1 else None) for call_id, count in Counter(call["id"] for call in message.tool_calls).items())
             elif isinstance(message, ToolMessage):
                 call_index = call_positions.pop(message.tool_call_id, None)
                 if call_index is not None and call_index in retained_positions and index in retained_positions:
diff --git a/backend/tests/test_subagent_context_snapshot.py b/backend/tests/test_subagent_context_snapshot.py
index 2972a1aa18..db3cfbddf4 100644
--- a/backend/tests/test_subagent_context_snapshot.py
+++ b/backend/tests/test_subagent_context_snapshot.py
@@ -371,6 +371,51 @@ def test_hidden_tool_frames_do_not_complete_visible_calls(hidden_part, tool_name
     assert ("TOOL_RESULT" in snapshot.content_json) is (hidden_part == "call")
 
 
+@pytest.mark.parametrize("result_count", [1, 2])
+@pytest.mark.parametrize("same_name", [False, True])
+def test_snapshot_omits_ambiguous_calls_without_losing_other_history(result_count, same_name):
+    messages = [
+        AIMessage(content="", tool_calls=[{"name": "bash", "args": {"input": "EARLIER_COMPLETED"}, "id": "reused"}]),
+        ToolMessage(content="Earlier result", tool_call_id="reused"),
+        AIMessage(
+            content="Keep this explanation",
+            tool_calls=[
+                {"name": "bash", "args": {"input": "AMBIGUOUS_FIRST"}, "id": "reused"},
+                {"name": "bash" if same_name else "task", "args": {"input": "AMBIGUOUS_SECOND"}, "id": "reused"},
+                {"name": "read_file", "args": {"path": "UNIQUE_COMPLETED"}, "id": "unique"},
+            ],
+        ),
+        *[ToolMessage(content=f"Observed result {index}", tool_call_id="reused") for index in range(result_count)],
+        ToolMessage(content="Unique result", tool_call_id="unique"),
+    ]
+    before = [message.model_dump() for message in messages]
+    snapshot = ParentContextSnapshot.from_state({"messages": messages})
+    assert "AMBIGUOUS" not in snapshot.content_json
+    assert "EARLIER_COMPLETED" in snapshot.content_json
+    assert "UNIQUE_COMPLETED" in snapshot.content_json
+    assert "Keep this explanation" in snapshot.content_json
+    assert "Observed result 0" in snapshot.content_json
+    assert [message.model_dump() for message in messages] == before
+
+
+def test_ambiguous_calls_do_not_reassign_results_to_an_earlier_pending_call():
+    messages = [
+        AIMessage(content="", tool_calls=[{"name": "task", "args": {"input": "EARLIER_PENDING"}, "id": "reused"}]),
+        AIMessage(
+            content="Current history",
+            tool_calls=[
+                {"name": "bash", "args": {"input": "AMBIGUOUS_FIRST"}, "id": "reused"},
+                {"name": "task", "args": {"input": "AMBIGUOUS_SECOND"}, "id": "reused"},
+            ],
+        ),
+        ToolMessage(content="Current result", tool_call_id="reused"),
+    ]
+    snapshot = ParentContextSnapshot.from_state({"messages": messages})
+    assert "EARLIER_PENDING" not in snapshot.content_json
+    assert "AMBIGUOUS" not in snapshot.content_json
+    assert "Current result" in snapshot.content_json
+
+
 @pytest.mark.parametrize("tool_name", ["task", "batch_task", "write_file", "bash", "custom_lookup"])
 @pytest.mark.parametrize("result_state", ["pending", "success", "error", "hidden"])
 def test_snapshot_keeps_only_result_paired_calls_in_mixed_dispatch(tool_name, result_state):
diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py
index 061529c320..dd0d43ddb6 100644
--- a/backend/tests/test_task_tool_core_logic.py
+++ b/backend/tests/test_task_tool_core_logic.py
@@ -1098,6 +1098,49 @@ def test_task_tool_context_mode_captures_dispatch_time_history(monkeypatch, cont
         assert captured.get("context_snapshot") is None
 
 
+@pytest.mark.parametrize("duplicate_id", [False, True])
+def test_task_dispatch_snapshot_does_not_include_unfinished_ambiguous_calls(monkeypatch, duplicate_id):
+    from langchain_core.messages import AIMessage, HumanMessage
+
+    runtime = _make_runtime()
+    runtime.state["messages"] = [
+        HumanMessage(content="Keep the user requirement"),
+        AIMessage(
+            content="",
+            tool_calls=[
+                {"name": "bash", "args": {"command": "COMPLETED_COMMAND"}, "id": "finished"},
+                {"name": "task", "args": {"prompt": "UNFINISHED_DELEGATION"}, "id": "finished" if duplicate_id else "pending"},
+            ],
+        ),
+        ToolMessage(content="Observed command result", tool_call_id="finished", name="bash"),
+    ]
+    before = [message.model_dump() for message in runtime.state["messages"]]
+    captured = {}
+
+    class DummyExecutor:
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+
+        def execute_async(self, prompt, task_id=None):
+            return task_id
+
+    monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
+    monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
+    monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
+    monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"))
+    monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda event: None)
+    monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+    result = _run_task_tool(runtime=runtime, prompt="Current delegated task", subagent_type="general-purpose", tool_call_id="current-dispatch", context_mode="snapshot")
+    assert _task_tool_message(result).additional_kwargs[SUBAGENT_STATUS_KEY] == "completed"
+    snapshot = captured["context_snapshot"]
+    assert "UNFINISHED_DELEGATION" not in snapshot.content_json
+    assert ("COMPLETED_COMMAND" in snapshot.content_json) is (not duplicate_id)
+    assert "Keep the user requirement" in snapshot.content_json
+    assert "Observed command result" in snapshot.content_json
+    assert [message.model_dump() for message in runtime.state["messages"]] == before
+
+
 def test_task_tool_context_mode_schema_rejects_unknown_mode():
     from pydantic import ValidationError