github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-31 15:40:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(web): preserve tool-result identity when provider call IDs repeat (#7672)

Browse Source 
* fix(web): preserve independent results when tool call IDs repeat

* fix(web): preserve leading results and idempotent result replay
This commit is contained in:
Lucien 2026-10-31 08:41:07 +08:00 committed by GitHub
parent 5d2749b05b
commit 178a9ab458
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 763 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@ -3267,6 +3267,12 @@ when retries are exhausted. Turn Live off and on to start a fresh retry budget.
**Visible Tool-Run Completion**: For interactive turns, DeerFlow retries an empty post-tool final response once, then surfaces a visible error instead of reporting a silent successful run.
**Run History**: Image-only user input is recorded once per run, even when the agent makes multiple model calls. Its media content remains in history without requiring a text summary.
Web history keeps separate tool results in their original positions when a
provider reuses a tool-call ID across turns or runs. Copies of the same result
are matched by their message ID. Legacy results without message IDs are kept
separately because their provider ID alone cannot identify an invocation.
A result delivered before its assistant still resolves when that call is unique
in the turn. Rebuilt results from the same current invocation keep one history entry.
### Reading a Referenced Conversation

22
backend/docs/runtime-guidance-details.md
View File

@ -45,7 +45,7 @@ once the feed's loaded page window no longer reaches back to it.
store already assigned, keyed by
`runtime/events/message_identity.py::message_identity` — the backend half of
the identity rule `frontend/src/core/threads/hooks.ts::messageIdentity` applies
(tool messages by `tool_call_id`; `X` / `X__user` human copies collapse to
(tool results by their own message ID; `X` / `X__user` human copies collapse to
one). The two halves must stay in sync: a mismatch is silent, degrading
placement rather than raising. `runs/worker.py::_MessageSeqStamper` attaches
the result as `additional_kwargs.deerflow_seq` on root `values` frames only —
@ -74,6 +74,26 @@ fetch-and-decode of every message row's tool outputs on long threads.
client input, because a welded-in seq goes stale when a fork re-seeds the feed
(#4380).
`RunJournal` assigns missing ToolMessage IDs on the returned message object
before serializing callback results, including `Command` message updates, so
the graph checkpoint and feed retain the same ID. Middleware results receive
the same treatment when reconciled. Provider `tool_call_id` values may repeat
within a run and across runs; they remain protocol correlation data, never
thread-global result identities. Legacy results without message IDs have no
dedupe identity or seq lookup rather than aliasing unrelated invocations.
For a registered lead-agent call, the journal remembers the captured result ID
until a new canonical assistant response emits that call ID again. Reconciliation
restores the remembered ID on a copy serialized before its first capture, so
the copy does not append another result or consume a sequence. An explicit
message ID is preserved. A later invocation starts fresh even if its provider
ID and output content are identical; duplicate callbacks for the same LLM run
do not reset this bookkeeping. Callbacks without a registered lead call still
assign distinct IDs to their returned objects. This cache is journal-local and
is cleared on detach, not a fallback identity for historical rows.
The web execution timeline and write-file preview resolve a provider call ID
only inside its originating assistant-message occurrence; UI keys include that
occurrence while outgoing protocol IDs remain unchanged.
**Run delivery receipts** (`runtime/journal.py` + `runs/worker.py`):
`RunJournal` records each non-empty artifact update once per tool `Command` for
the terminal `run.delivery` event. When a command contains multiple messages, a

2
backend/packages/harness/deerflow/runtime/AGENTS.md
View File

@ -83,7 +83,7 @@ without querying the repository or replacing the control input.
Before changing checkpoint modes, caches, replay lookup, state replacement or
rollback, read the [checkpoint contracts](../../../../docs/runtime-guidance-details.md#checkpoint-channel-modes-and-replay).
**Message sequence placement:** Keep backend and frontend message identity rules aligned. Details: `backend/docs/runtime-guidance-details.md`.
**Message sequence placement:** Keep backend and frontend message identity rules aligned. Tool results use their own message IDs, assigned on missing-ID callback results before persistence; provider call IDs may repeat. Restore pre-capture ID-less copies from the current lead-call result ID; reset it on a new canonical call, not a usage replay. Keep ID-less legacy rows independent. Details: `backend/docs/runtime-guidance-details.md`.
**Thread message cursors:** `list_messages` applies both exclusive bounds before `limit`, paging forward whenever `after_seq` is supplied.

8
backend/packages/harness/deerflow/runtime/events/message_identity.py
View File

@ -8,8 +8,8 @@ stay in sync: a mismatch is silent, degrading placement rather than raising.
Two normalizations matter:
* a ``ToolMessage`` is identified by its ``tool_call_id``, not its own id —
that is the id both sides can always resolve;
* a ``ToolMessage`` is identified by its own message id, assigned by the journal
before persistence; provider ``tool_call_id`` values can repeat across turns;
* ``DynamicContextMiddleware`` re-keys the submitted user turn from ``X`` to
``X__user`` (giving ``X`` to the injected reminder), so the two human copies
must collapse to one identity.
@ -35,10 +35,6 @@ def message_identity(message: Mapping[str, Any]) -> str | None:
*message* is a serialized message mapping (as stored in ``run_events``
content or carried in a checkpoint ``values`` frame), not a ``BaseMessage``.
"""
tool_call_id = message.get("tool_call_id")
if isinstance(tool_call_id, str) and tool_call_id:
return f"tool:{tool_call_id}"
message_id = message.get("id")
if not isinstance(message_id, str) or not message_id:
return None

23
backend/packages/harness/deerflow/runtime/journal.py
View File

@ -28,7 +28,7 @@ from dataclasses import dataclass
from datetime import UTC, datetime
from functools import partial
from typing import Any, cast
from uuid import UUID
from uuid import UUID, uuid4
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolMessage, messages_from_dict
@ -420,6 +420,7 @@ class RunJournal(BaseCallbackHandler):
self._llm_call_index = 0
self._seen_llm_starts: set[str] = set() # langchain run_ids that fired on_chat_model_start
self._current_run_tool_call_names: dict[str, str] = {}
self._current_tool_result_ids: dict[str, str] = {}
self._active_tool_names: dict[str, str] = {}
self._persisted_ai_message_identities: set[str] = set()
self._persisted_tool_message_identities: set[str] = set()
@ -842,9 +843,6 @@ class RunJournal(BaseCallbackHandler):
@staticmethod
def _message_identity(message: BaseMessage) -> str | None:
tool_call_id = getattr(message, "tool_call_id", None)
if isinstance(tool_call_id, str) and tool_call_id:
return f"tool:{tool_call_id}"
message_id = getattr(message, "id", None)
if isinstance(message_id, str) and message_id:
return f"message:{message_id}"
@ -883,8 +881,16 @@ class RunJournal(BaseCallbackHandler):
continue
name = self._tool_call_value(tool_call, "name")
self._current_run_tool_call_names[tool_call_id] = str(name or "")
# A newly emitted assistant call starts a distinct occurrence,
# even when the provider reuses an earlier call ID in this run.
self._current_tool_result_ids.pop(tool_call_id, None)
def _persist_tool_result_message(self, message: BaseMessage) -> None:
# Tool callbacks run before the graph reducer assigns missing ids.
# Assign on the returned object so the feed and checkpoint share the
# same result identity even when a provider reuses its tool_call_id.
if isinstance(message, ToolMessage) and not message.id:
message.id = str(uuid4())
self._put(
event_type=LLM_TOOL_RESULT_EVENT.event_type,
category=LLM_TOOL_RESULT_EVENT.category,
@ -893,6 +899,8 @@ class RunJournal(BaseCallbackHandler):
identity = self._message_identity(message)
if identity:
self._persisted_tool_message_identities.add(identity)
if isinstance(message, ToolMessage) and message.tool_call_id in self._current_run_tool_call_names and message.id:
self._current_tool_result_ids[message.tool_call_id] = message.id
self._record_message_summary(message)
def record_interrupted_ai_responses(self, messages: Iterable[AIMessage]) -> int:
@ -959,8 +967,12 @@ class RunJournal(BaseCallbackHandler):
# re-attributed here.
if self._current_run_tool_call_names.get(tool_call_id) is None:
return False
if not message.id:
# A copy serialized before the first capture still lacks its ID.
# Restore that occurrence's ID before checking for a replay.
message.id = self._current_tool_result_ids.get(tool_call_id)
identity = self._message_identity(message)
return identity is not None and identity not in self._persisted_tool_message_identities
return identity is None or identity not in self._persisted_tool_message_identities
def _reconcile_tool_messages(self, messages: Iterable[Any]) -> None:
for message in messages:
@ -1823,6 +1835,7 @@ class RunJournal(BaseCallbackHandler):
self._llm_start_times.clear()
self._seen_llm_starts.clear()
self._current_run_tool_call_names.clear()
self._current_tool_result_ids.clear()
self._persisted_ai_message_identities.clear()
self._persisted_tool_message_identities.clear()
self._produced_artifacts.clear()

16
backend/tests/test_run_event_store.py
View File

@ -1568,7 +1568,7 @@ class TestGetMessageSeqs:
assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1}
@pytest.mark.anyio
async def test_a_tool_message_is_identified_by_its_tool_call_id(self, store):
async def test_a_tool_message_is_identified_by_its_message_id(self, store):
await store.put(
thread_id="t1",
run_id="r1",
@ -1577,7 +1577,7 @@ class TestGetMessageSeqs:
content={"type": "tool", "id": "lc-abc", "tool_call_id": "call_1", "content": "OK"},
)
assert await store.get_message_seqs("t1", ["tool:call_1"]) == {"tool:call_1": 1}
assert await store.get_message_seqs("t1", ["message:lc-abc"]) == {"message:lc-abc": 1}
@pytest.mark.anyio
async def test_the_injected_user_suffix_collapses_to_one_identity(self, store):
@ -1698,12 +1698,12 @@ class TestGetMessageSeqs:
run_id="r1",
event_type="llm.tool.result",
category="message",
content={"type": "tool", "tool_call_id": "call_1", "content": "OK"},
content={"type": "tool", "id": "result-1", "tool_call_id": "call_1", "content": "OK"},
)
assert await s.get_message_seqs("t1", ["message:u1", "tool:call_1"]) == {
assert await s.get_message_seqs("t1", ["message:u1", "message:result-1"]) == {
"message:u1": 1,
"tool:call_1": 2,
"message:result-1": 2,
}
@pytest.mark.anyio
@ -1727,7 +1727,7 @@ class TestGetMessageSeqs:
run_id="r1",
event_type="llm.tool.result",
category="message",
content={"type": "tool", "tool_call_id": "call_1", "content": "OK"},
content={"type": "tool", "id": "result-1", "tool_call_id": "call_1", "content": "OK"},
)
await s.put(
thread_id="t1",
@ -1737,9 +1737,9 @@ class TestGetMessageSeqs:
content={"type": "human", "id": "ignored"},
)
assert await s.get_message_seqs("t1", ["message:u1", "tool:call_1", "message:ignored"]) == {
assert await s.get_message_seqs("t1", ["message:u1", "message:result-1", "message:ignored"]) == {
"message:u1": 1,
"tool:call_1": 2,
"message:result-1": 2,
}
finally:
await close_engine()

128
backend/tests/test_tool_result_identity.py Normal file
View File

@ -0,0 +1,128 @@
"""Tool invocation identity survives provider ID reuse and graph serialization."""
from uuid import uuid4
import pytest
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, LLMResult
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.types import Command
from deerflow.agents.thread_state import get_thread_state_schema
from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
from deerflow.runtime.events.message_identity import message_identity
from deerflow.runtime.events.message_seq import stamp_messages_with_seq
from deerflow.runtime.events.store.memory import MemoryRunEventStore
from deerflow.runtime.journal import RunJournal
def test_idless_legacy_results_do_not_claim_a_shared_identity():
assert message_identity({"type": "tool", "tool_call_id": "call_0"}) is None
@pytest.mark.anyio
@pytest.mark.parametrize("backend", ["memory", "jsonl", "db"])
@pytest.mark.parametrize("second_run", ["run-a", "run-b"])
async def test_reused_provider_ids_keep_distinct_seqs(backend, second_run, tmp_path):
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
from deerflow.runtime.events.store.db import DbRunEventStore
from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
if backend == "db":
await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'events.db'}", sqlite_dir=str(tmp_path))
store = DbRunEventStore(get_session_factory())
elif backend == "jsonl":
store = JsonlRunEventStore(base_dir=tmp_path / "events")
else:
store = MemoryRunEventStore()
first = {"type": "tool", "id": "result-first", "tool_call_id": "call_0", "content": "first"}
second = {**first, "id": "result-second", "content": "second"}
try:
for run_id, message in [("run-a", first), (second_run, second), (second_run, {**second, "content": "replayed"})]:
await store.put(thread_id="thread", run_id=run_id, event_type="llm.tool.result", category="message", content=message)
stamped = await stamp_messages_with_seq(store, "thread", [first, second])
assert [message["additional_kwargs"]["deerflow_seq"] for message in stamped] == [1, 2]
assert [message_identity(message) for message in stamped] == ["message:result-first", "message:result-second"]
finally:
if backend == "db":
await close_engine()
@pytest.mark.anyio
@pytest.mark.parametrize("command_result", [False, True])
@pytest.mark.parametrize("mode", ["full", "delta"])
async def test_tool_callback_ids_match_reloaded_checkpoint(command_result, mode):
"""Exercise real BaseTool callbacks, ToolNode, reducer and checkpointer."""
@tool
def lookup(value: str) -> str | Command:
"""Return the requested value."""
if command_result:
return Command(update={"messages": [ToolMessage(content=value, tool_call_id="call_0")]})
return value
builder = StateGraph(get_thread_state_schema(mode))
builder.add_node("tools", ToolNode([lookup]))
builder.set_entry_point("tools")
builder.set_finish_point("tools")
saver = InMemorySaver()
graph = builder.compile(checkpointer=saver)
accessor = CheckpointStateAccessor.bind(graph, saver, mode=mode)
store = MemoryRunEventStore()
journal = RunJournal("run-a", "thread", store)
config = {"configurable": {"thread_id": "thread"}, "callbacks": [journal]}
try:
for index in range(2):
await graph.ainvoke(
{"messages": [AIMessage(id=f"assistant-{index}", content="", tool_calls=[{"id": "call_0", "name": "lookup", "args": {"value": f"result-{index}"}}])]},
config,
)
await journal.flush()
events = await store.list_messages("thread")
feed_tools = [event["content"] for event in events if event["content"].get("type") == "tool"]
snapshot = await accessor.aget(config)
tools = [message for message in snapshot.values["messages"] if isinstance(message, ToolMessage)]
assert len(feed_tools) == len(tools) == 2
assert len({message["id"] for message in feed_tools}) == 2
assert all(message["id"] for message in feed_tools)
assert [message["id"] for message in feed_tools] == [message.id for message in tools]
stamped = await stamp_messages_with_seq(store, "thread", [message.model_dump() for message in tools])
assert [message["additional_kwargs"]["deerflow_seq"] for message in stamped] == [event["seq"] for event in events if event["content"].get("type") == "tool"]
finally:
await journal.close()
@pytest.mark.anyio
@pytest.mark.parametrize("with_id", [False, True])
@pytest.mark.parametrize("callback", [False, True])
@pytest.mark.parametrize("same_content", [False, True])
async def test_middleware_results_with_reused_call_id_are_reconciled_once(with_id, callback, same_content):
store = MemoryRunEventStore()
journal = RunJournal("run-a", "thread", store)
try:
for index in range(2):
assistant = AIMessage(id=f"assistant-{index}", content="", tool_calls=[{"id": "call_0", "name": "blocked", "args": {}}])
response = LLMResult(generations=[[ChatGeneration(message=assistant)]])
completion_id = uuid4()
journal.on_llm_end(response, run_id=completion_id, tags=["lead_agent"])
result = ToolMessage(id=f"result-{index}" if with_id else None, content="blocked" if same_content else f"blocked-{index}", tool_call_id="call_0")
# Capture the wire shape before either callback can assign an ID.
snapshot = result.model_dump()
if callback:
journal.on_tool_end(result, run_id=uuid4())
journal.on_chat_model_start({}, [[assistant, result]], run_id=uuid4(), tags=["lead_agent"])
# A duplicate completion callback must not reopen the occurrence.
journal.on_llm_end(response, run_id=completion_id, tags=["lead_agent"])
replay = ToolMessage.model_validate(snapshot)
journal.on_chat_model_start({}, [[assistant, replay]], run_id=uuid4(), tags=["lead_agent"])
assert replay.id == result.id
assert result.id
await journal.flush()
results = [event["content"] for event in await store.list_messages("thread") if event["content"].get("type") == "tool"]
assert [result["content"] for result in results] == (["blocked", "blocked"] if same_content else ["blocked-0", "blocked-1"])
assert len({result["id"] for result in results}) == 2
finally:
await journal.close()

4
frontend/src/AGENTS.md
View File

File diff suppressed because one or more lines are too long

6
frontend/src/components/workspace/artifacts/artifact-file-detail.tsx
View File

@ -185,7 +185,11 @@ export function ArtifactFileDetail({
if (!toolCallId) {
return undefined;
}
return findToolCallResult(toolCallId, thread.messages);
return findToolCallResult(
toolCallId,
thread.messages,
parsedWriteFile?.messageId,
);
})();
const artifactViewState = getArtifactViewState({
filepath: filepathFromProps,

64
frontend/src/components/workspace/messages/message-group.tsx
View File

@ -36,6 +36,7 @@ import {
} from "@/core/artifacts/utils";
import { useI18n } from "@/core/i18n/hooks";
import { isErrorToolMessage } from "@/core/messages/tool-error";
import { indexToolResults } from "@/core/messages/tool-result";
import { formatTokenCount } from "@/core/messages/usage";
import type { TokenDebugStep } from "@/core/messages/usage-model";
import {
@ -294,7 +295,7 @@ function MessageGroupComponent({
return (
<ToolCall
key={step.id}
key={step.occurrenceId}
{...step}
threadId={threadId}
isLast={options?.isLast}
@ -414,7 +415,7 @@ function MessageGroupComponent({
lastToolCallStep.messageId,
stepIndexByStep.get(lastToolCallStep) ?? -1,
)}
<FlipDisplay uniqueKey={lastToolCallStep.id ?? ""}>
<FlipDisplay uniqueKey={lastToolCallStep.occurrenceId}>
{renderToolCall(lastToolCallStep, { isLast: true })}
</FlipDisplay>
{afterLastToolCallAssistantTextSteps
@ -1210,6 +1211,7 @@ interface CoTReasoningStep extends GenericCoTStep<"reasoning"> {
}
interface CoTToolCallStep extends GenericCoTStep<"toolCall"> {
occurrenceId: string;
name: string;
args: Record<string, unknown>;
result?: string;
@ -1235,31 +1237,32 @@ function indexToolCallData(messages: Message[]) {
const toolCallResults = new Map<string, string>();
const browserViews = new Map<string, BrowserViewMeta>();
const resultMessages = new Map<string, Extract<Message, { type: "tool" }>>();
for (const message of messages) {
if (message.type !== "tool" || !message.tool_call_id) {
continue;
}
const toolCallId = message.tool_call_id;
if (!resultMessages.has(toolCallId))
resultMessages.set(toolCallId, message);
if (!toolCallResults.has(toolCallId)) {
const result = extractTextFromMessage(message);
if (result) {
toolCallResults.set(toolCallId, result);
resultMessages.set(toolCallId, message);
for (const [ownerIndex, results] of indexToolResults(messages)) {
for (const message of results) {
if (message.type !== "tool" || !message.tool_call_id) {
continue;
}
}
if (!browserViews.has(toolCallId)) {
const browserView = (
message.additional_kwargs as
| { browser_view?: BrowserViewMeta }
| undefined
)?.browser_view;
if (browserView && typeof browserView.screenshot === "string") {
browserViews.set(toolCallId, browserView);
const toolCallId = `${ownerIndex}:${message.tool_call_id}`;
if (!resultMessages.has(toolCallId))
resultMessages.set(toolCallId, message);
if (!toolCallResults.has(toolCallId)) {
const result = extractTextFromMessage(message);
if (result) {
toolCallResults.set(toolCallId, result);
resultMessages.set(toolCallId, message);
}
}
if (!browserViews.has(toolCallId)) {
const browserView = (
message.additional_kwargs as
| { browser_view?: BrowserViewMeta }
| undefined
)?.browser_view;
if (browserView && typeof browserView.screenshot === "string") {
browserViews.set(toolCallId, browserView);
}
}
}
}
@ -1316,6 +1319,10 @@ function convertToSteps(
}
const step: CoTToolCallStep = {
id: tool_call.id,
occurrenceId: JSON.stringify([
message.id ?? `ai-${messageIndex}`,
tool_call.id,
]),
messageId: message.id,
type: "toolCall",
name: tool_call.name,
@ -1325,8 +1332,9 @@ function convertToSteps(
};
const toolCallId = tool_call.id;
if (toolCallId) {
const toolCallResult = toolCallResults.get(toolCallId);
step.resultMessage = resultMessages.get(toolCallId);
const occurrence = `${messageIndex}:${toolCallId}`;
const toolCallResult = toolCallResults.get(occurrence);
step.resultMessage = resultMessages.get(occurrence);
// Generic details preserve received text; specialized tools retain their parsing.
if (toolCallResult && getToolCallKind(tool_call.name) !== "generic") {
try {
@ -1336,7 +1344,7 @@ function convertToSteps(
step.result = toolCallResult;
}
}
step.browserView = browserViews.get(toolCallId);
step.browserView = browserViews.get(occurrence);
const artifacts = artifactsByToolCallId.get(toolCallId);
if (artifacts) {
step.artifacts = artifacts;

7
frontend/src/core/artifacts/preview.ts
View File

@ -1,3 +1,5 @@
import { indexToolResults } from "../messages/tool-result";
export function getTabularDelimiter(
language: string | null,
): "," | "\t" | null {
@ -98,8 +100,9 @@ export function buildWriteFileDraftContent({
let draft = "";
let hasDraft = false;
const toolResults = indexToolResults(messages);
for (const message of messages) {
for (const [messageIndex, message] of messages.entries()) {
if (message.type !== "ai") {
continue;
}
@ -116,7 +119,7 @@ export function buildWriteFileDraftContent({
const toolCallId = toolCall.id;
const toolResult = toolCallId
? findToolResult(toolCallId, messages)
? findToolResult(toolCallId, toolResults.get(messageIndex) ?? [])
: undefined;
const isSelected =
toolCallId === target.toolCallId &&

7
frontend/src/core/messages/derived-state.ts
View File

@ -9,13 +9,6 @@ import {
} from "./utils";
function messageStableKey(message: Message) {
if (
message.type === "tool" &&
typeof message.tool_call_id === "string" &&
message.tool_call_id.length > 0
) {
return `tool:${message.tool_call_id}`;
}
if (typeof message.id === "string" && message.id.length > 0) {
return `message:${message.id}`;
}

58
frontend/src/core/messages/tool-result.ts Normal file
View File

@ -0,0 +1,58 @@
type ToolResultMessage = {
type?: string;
tool_call_id?: string;
tool_calls?: Array<{ id?: string }>;
};
/** Pair results with assistant occurrences, preserving unique leading SDK results. */
export function indexToolResults<T extends ToolResultMessage>(
messages: readonly T[],
) {
const results = new Map<number, T[]>();
let groupStart = 0;
for (let end = 0; end <= messages.length; end += 1) {
if (end < messages.length && messages[end]?.type !== "human") continue;
const uniqueOwners = new Map<string, number>();
const previousAssistant = new Map<number, number>();
let assistantIndex = -1;
for (let index = groupStart; index < end; index += 1) {
const message = messages[index]!;
if (message.type !== "ai") continue;
previousAssistant.set(index, assistantIndex);
assistantIndex = index;
for (const call of message.tool_calls ?? []) {
if (!call.id) continue;
uniqueOwners.set(call.id, uniqueOwners.has(call.id) ? -1 : index);
}
}
assistantIndex = -1;
let assistantCalls = new Set<string | undefined>();
for (let index = groupStart; index < end; index += 1) {
const message = messages[index]!;
if (message.type === "ai") {
assistantIndex = index;
assistantCalls = new Set(message.tool_calls?.map((call) => call.id));
}
if (message.type !== "tool" || !message.tool_call_id) continue;
const uniqueOwner = uniqueOwners.get(message.tool_call_id);
// Only the next assistant can own a leading result. Never bridge a human
// or another assistant, and never guess forward when a call ID repeats.
const owner =
uniqueOwner !== undefined &&
uniqueOwner > index &&
previousAssistant.get(uniqueOwner)! < index
? uniqueOwner
: assistantCalls.has(message.tool_call_id)
? assistantIndex
: -1;
if (owner < 0) continue;
const owned = results.get(owner);
if (owned) owned.push(message);
else results.set(owner, [message]);
}
groupStart = end + 1;
}
return results;
}

18
frontend/src/core/messages/utils.ts
View File

@ -13,6 +13,7 @@ import {
} from "@/core/streamdown/fences";
import { modelControlDisplayText } from "./model-control";
import { indexToolResults } from "./tool-result";
interface GenericMessageGroup<T = string> {
type: T;
@ -1113,8 +1114,21 @@ export function hasSubagent(message: AIMessage) {
return false;
}
export function findToolCallResult(toolCallId: string, messages: Message[]) {
for (const message of messages) {
export function findToolCallResult(
toolCallId: string,
messages: Message[],
assistantMessageId?: string,
) {
const start = assistantMessageId
? messages.findIndex(
(message) => message.type === "ai" && message.id === assistantMessageId,
)
: -1;
if (assistantMessageId && start < 0) return undefined;
const candidates = assistantMessageId
? (indexToolResults(messages).get(start) ?? [])
: messages;
for (const message of candidates) {
if (message.type === "tool" && message.tool_call_id === toolCallId) {
const content = extractTextFromMessage(message);
if (content) {

12
frontend/src/core/threads/hooks.ts
View File

@ -22,6 +22,7 @@ import { useI18n } from "../i18n/hooks";
import type { Locale } from "../i18n/locale";
import { getMessageRunId } from "../messages/run-duration";
import { isErrorToolMessage } from "../messages/tool-error";
import { indexToolResults } from "../messages/tool-result";
import {
extractContentFromMessage,
hasToolCalls,
@ -163,19 +164,24 @@ export function getToolResultStatus(
toolName: string,
): ToolResultStatus {
let latestToolCallId: string | undefined;
for (const message of messages) {
let latestToolCallIndex = -1;
for (const [messageIndex, message] of messages.entries()) {
if (message.type !== "ai") {
continue;
}
for (const toolCall of message.tool_calls ?? []) {
if (toolCall.name === toolName && toolCall.id) {
latestToolCallId = toolCall.id;
latestToolCallIndex = messageIndex;
}
}
}
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index];
const results = latestToolCallId
? (indexToolResults(messages).get(latestToolCallIndex) ?? [])
: messages;
for (let index = results.length - 1; index >= 0; index -= 1) {
const message = results[index];
if (message?.type !== "tool") {
continue;
}

10
frontend/src/core/threads/message-order.ts
View File

@ -39,13 +39,9 @@ export const MESSAGE_SEQ_KEY = "deerflow_seq";
const INJECTED_USER_MESSAGE_ID_SUFFIX = "__user";
export function messageIdentity(message: Message): string | undefined {
if (
"tool_call_id" in message &&
typeof message.tool_call_id === "string" &&
message.tool_call_id.length > 0
) {
return `tool:${message.tool_call_id}`;
}
// Providers can reuse tool_call_id across runs and assistant turns. The
// journal assigns result ids before persistence so feed/checkpoint copies
// share one identity. Legacy id-less results must not collapse by call id.
if (typeof message.id === "string" && message.id.length > 0) {
// DynamicContextMiddleware replaces the submitted HumanMessage(id=X) with
// a hidden SystemMessage(id=X) and the real HumanMessage(id=X__user).

47
frontend/tests/unit/components/workspace/artifacts/artifact-file-detail.dom.test.tsx
View File

@ -1,9 +1,11 @@
import type { Message } from "@langchain/langgraph-sdk";
import { afterEach, beforeEach, describe, expect, it, rs } from "@rstest/core";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import { cleanup, render } from "@testing-library/react";
import type { PropsWithChildren } from "react";
const mocks = rs.hoisted(() => ({
messages: [] as Message[],
artifactContent: {
content: undefined as string | undefined,
url: undefined as string | undefined,
@ -23,7 +25,10 @@ rs.mock("@/core/artifacts/hooks", () => ({
}));
rs.mock("@/components/workspace/messages/context", () => ({
useThread: () => ({ thread: { isLoading: false }, isMock: false }),
useThread: () => ({
thread: { isLoading: false, messages: mocks.messages },
isMock: false,
}),
}));
rs.mock("@/components/workspace/artifacts/context", () => ({
@ -90,10 +95,50 @@ function renderDetail(filepath: string) {
}
beforeEach(() => {
mocks.messages = [];
mocks.artifactContent.content = undefined;
mocks.artifactContent.sha256 = undefined;
});
describe("ArtifactFileDetail leading tabular write results", () => {
it.each(["OK", "Error: failed", undefined])(
"enables table preview only after this call succeeds: %s",
(result) => {
const path = "/mnt/user-data/outputs/table.csv";
mocks.messages = [
...(result === undefined
? []
: [
{
type: "tool",
id: "result",
tool_call_id: "call_0",
content: result,
} as Message,
]),
{
type: "ai",
id: "owner",
content: "",
tool_calls: [
{
id: "call_0",
name: "write_file",
args: { path, content: "name,value\nrow,1" },
},
],
},
];
const view = renderDetail(
`write-file:${path}?message_id=owner&tool_call_id=call_0`,
);
expect(view.queryByLabelText("Table preview") !== null).toBe(
result === "OK",
);
},
);
});
describe("ArtifactFileDetail extensionless source files", () => {
it.each([
["Dockerfile", "dockerfile", "FROM python:3.13\n"],

31
frontend/tests/unit/components/workspace/messages/message-group.test.ts
View File

@ -38,6 +38,37 @@ afterEach(() => {
});
describe("MessageGroup", () => {
it("renders each assistant turn's result when provider tool-call ids repeat", () => {
const messages: Message[] = ["first", "second"].flatMap((turn) => [
{
id: `assistant-${turn}`,
type: "ai",
content: "",
tool_calls: [
{
id: "call_0",
name: "web_fetch",
args: { url: `https://example.com/${turn}` },
},
],
} as Message,
{
id: `result-${turn}`,
type: "tool",
name: "web_fetch",
tool_call_id: "call_0",
content: `# ${turn} fetched title\n\n${turn} result`,
} as Message,
]);
expect(renderGroup(messages.slice(0, 2))).toContain("first fetched title");
// Earlier tool steps are collapsed by default; the current card must show
// its own result rather than borrowing the first call's fetched title.
const html = renderGroup(messages);
expect(html).toContain("second fetched title");
expect(html).not.toContain("first fetched title");
});
it("resolves relative artifact links in chain-of-thought assistant text", () => {
const html = renderGroup(
[

82
frontend/tests/unit/core/artifacts/preview.test.ts
View File

@ -11,6 +11,88 @@ import {
const ARTIFACT_PATH = "/artifact-fixtures/report.html";
const UNSUPPORTED_ARTIFACT_PATH = "/artifact-fixtures/data.csv";
test.each(["OK", "Error: failed", undefined])(
"write drafts respect a unique call's leading result: %s",
(result) => {
const owner = {
type: "ai",
id: "owner",
tool_calls: [
{
id: "call_0",
name: "write_file",
args: { path: ARTIFACT_PATH, content: "draft" },
},
],
};
const messages = [
...(result === undefined
? []
: [{ type: "tool", tool_call_id: "call_0", content: result }]),
owner,
];
expect(
buildWriteFileDraftContent({
filepath: `write-file:${ARTIFACT_PATH}?message_id=owner&tool_call_id=call_0`,
messages,
}),
).toBe(result === "Error: failed" ? undefined : "draft");
},
);
test("write drafts use their own assistant occurrence when call ids repeat", () => {
const messages = [
{
type: "ai",
id: "a1",
tool_calls: [
{
id: "call_0",
name: "write_file",
args: { path: ARTIFACT_PATH, content: "first" },
},
],
},
{ type: "tool", id: "t1", tool_call_id: "call_0", content: "OK" },
{
type: "ai",
id: "a2",
tool_calls: [
{
id: "call_0",
name: "write_file",
args: { path: ARTIFACT_PATH, content: "failed" },
},
],
},
{
type: "tool",
id: "t2",
tool_call_id: "call_0",
content: "Error: failed",
},
{
type: "ai",
id: "a3",
tool_calls: [
{
id: "call_0",
name: "write_file",
args: { path: ARTIFACT_PATH, content: "pending" },
},
],
},
];
const draft = (messageId: string) =>
buildWriteFileDraftContent({
filepath: `write-file:${ARTIFACT_PATH}?message_id=${messageId}&tool_call_id=call_0`,
messages,
});
expect(draft("a1")).toBe("first");
expect(draft("a2")).toBeUndefined();
expect(draft("a3")).toBe("pending");
});
test("allows in-progress write artifacts to render a throttled preview", () => {
expect(
getArtifactViewState({

42
frontend/tests/unit/core/messages/derived-state.test.ts
View File

@ -17,6 +17,48 @@ function message(type: Message["type"], id: string, content: string): Message {
}
describe("incremental message derivation", () => {
it("reuses result copies by message id and keeps ID-less results distinct", () => {
const assistant = {
type: "ai",
id: "a1",
content: "",
tool_calls: [{ id: "call_0", name: "read_file", args: {} }],
} as Message;
for (const id of ["result-1", undefined]) {
const result = {
type: "tool",
...(id ? { id } : {}),
tool_call_id: "call_0",
content: "same output",
} as Message;
const initial = deriveStableMessageGroups(
[assistant, result],
false,
[],
false,
);
const replay = deriveStableMessageGroups(
[assistant, { ...result }],
false,
initial,
false,
);
if (id) {
expect(replay[0]).toBe(initial[0]);
const other = deriveStableMessageGroups(
[assistant, { ...result, id: "result-2" }],
false,
initial,
false,
);
expect(other[0]).not.toBe(initial[0]);
expect(other[0]?.messages.at(-1)?.id).toBe("result-2");
} else {
expect(replay[0]).not.toBe(initial[0]);
}
}
});
it("reuses completed groups when only the streaming turn changes", () => {
const messages = Array.from({ length: 1_000 }, (_, turn) => [
message("human", `h-${turn}`, `question ${turn}`),

77
frontend/tests/unit/core/messages/utils.test.ts
View File

@ -6,6 +6,7 @@ import {
extractContentFromMessage,
extractTextFromMessage,
extractReasoningContentFromMessage,
findToolCallResult,
getBranchableAssistantGroupIds,
getLatestEditableTurn,
getMessageCopyData,
@ -39,6 +40,82 @@ function aiMessage(content: string): Message {
} as Message;
}
test.each(["OK", "Error: failed"])(
"selected unique calls resolve a leading result: %s",
(content) => {
const messages = [
{ id: "result", type: "tool", tool_call_id: "call_0", content },
{
id: "owner",
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "write_file", args: {} }],
},
] as Message[];
expect(findToolCallResult("call_0", messages, "owner")).toBe(content);
expect(findToolCallResult("call_0", messages, "missing")).toBeUndefined();
expect(
findToolCallResult("call_0", messages.slice(1), "owner"),
).toBeUndefined();
const nextTurn = [
messages[0]!,
{ type: "human", content: "next" } as Message,
messages[1]!,
];
expect(findToolCallResult("call_0", nextTurn, "owner")).toBeUndefined();
},
);
test("tool result lookup stays within the selected assistant occurrence", () => {
const assistant = (id: string): Message => ({
id,
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "write_file", args: {} }],
});
const messages = [
assistant("a1"),
{ id: "t1", type: "tool", tool_call_id: "call_0", content: "OK" },
assistant("a2"),
{
id: "t2",
type: "tool",
tool_call_id: "call_0",
content: "Error: failed",
},
assistant("a3"),
] as Message[];
expect(findToolCallResult("call_0", messages, "a1")).toBe("OK");
expect(findToolCallResult("call_0", messages, "a2")).toBe("Error: failed");
expect(findToolCallResult("call_0", messages, "a3")).toBeUndefined();
expect(findToolCallResult("call_0", messages, "unloaded")).toBeUndefined();
});
test.each([false, true])(
"an unrelated assistant fences result lookup (leading=%s)",
(leading) => {
const owner = {
id: "owner",
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "write_file", args: {} }],
} as Message;
const other = {
id: "other",
type: "ai",
content: "An unrelated answer",
} as Message;
const result = {
type: "tool",
tool_call_id: "call_0",
content: "OK",
} as Message;
const messages = leading ? [result, other, owner] : [owner, other, result];
expect(findToolCallResult("call_0", messages, "owner")).toBeUndefined();
expect(findToolCallResult("call_0", messages, "other")).toBeUndefined();
},
);
test("aggregates token usage messages once per assistant turn", () => {
const messages = [
{

20
frontend/tests/unit/core/threads/message-merge.test.ts
View File

@ -262,15 +262,15 @@ test("mergeMessages keeps live-only messages between shared anchors in place", (
]);
});
test("mergeMessages deduplicates tool messages by tool_call_id", () => {
test("mergeMessages deduplicates copies of the same tool result by message id", () => {
const oldTool = {
id: "tool-message-old",
id: "tool-message",
type: "tool",
tool_call_id: "call-1",
content: "old",
} as Message;
const liveTool = {
id: "tool-message-live",
id: "tool-message",
type: "tool",
tool_call_id: "call-1",
content: "live",
@ -1674,7 +1674,7 @@ test("local turn order keeps an existing clarification card with its original tu
const baselineIdentities = new Set([
"message:previous-human",
"message:previous-answer",
"tool:clarification-call",
"message:clarification-card",
]);
// A live checkpoint tail can be woven after the newly persisted human
@ -1738,7 +1738,7 @@ test("local turn order preserves relative order across displaced message sources
currentStep,
],
new Set(["message:previous-human", "message:baseline-answer"]),
new Set(["tool:history-only-call"]),
new Set(["message:history-only-card"]),
),
).toEqual([
previousHuman,
@ -1793,7 +1793,7 @@ test("current turn run id derivation selects only visible unconfirmed steps", ()
confirmedHistoryStep,
],
new Set(["message:baseline-step"]),
new Set(["tool:confirmed-history-call"]),
new Set(["message:confirmed-history-step"]),
),
).toEqual(new Set(["run-current"]));
expect(
@ -1823,7 +1823,7 @@ test("history-confirmed live cards are not derived as current-run steps", () =>
run_id: "run-previous",
} as Message;
const baselineIdentities = new Set(["message:previous-human"]);
const historyIdentities = new Set(["tool:old-history-call"]);
const historyIdentities = new Set(["message:old-history-card"]);
const currentTurnRunIds = getCurrentTurnRunIds(
[oldHistoryCard],
baselineIdentities,
@ -1861,7 +1861,7 @@ test("local turn order repairs a displaced suffix when the human is first", () =
restoreLocalTurnMessageOrder(
[currentHuman, oldHistoryCard],
new Set(),
new Set(["tool:old-history-call"]),
new Set(["message:old-history-card"]),
),
).toEqual([oldHistoryCard, currentHuman]);
});
@ -1986,7 +1986,7 @@ test("local turn order keeps the current run's persisted steps after the new hum
"message:previous-human",
"message:current-human",
"message:current-step",
"tool:call-current",
"message:current-tool",
]);
// The human already at its correct position, followed by its own steps.
@ -2084,7 +2084,7 @@ test("local turn order restores a clarification card that only canonical history
const historyIdentities = new Set([
"message:previous-human",
"message:previous-answer",
"tool:clarification-call",
"message:clarification-card",
]);
// The merged list has the card woven after the new human (from history).

59
frontend/tests/unit/core/threads/message-order.test.ts
View File

@ -75,7 +75,7 @@ function idsOf(messages: Message[]): (string | undefined)[] {
return messages.map((message) => message.id);
}
test("messageIdentity collapses X and X__user and keys tools by tool_call_id", () => {
test("messageIdentity collapses X and X__user and keys tools by message id", () => {
expect(messageIdentity(msg("req-1", "human", "q"))).toBe("message:req-1");
expect(messageIdentity(msg("req-1__user", "human", "q"))).toBe(
"message:req-1",
@ -92,7 +92,7 @@ test("messageIdentity collapses X and X__user and keys tools by tool_call_id", (
content: "result",
tool_call_id: "call-1",
} as Message),
).toBe("tool:call-1");
).toBe("message:tool-own-id");
});
test("trustedMessageSeq accepts only positive safe integers", () => {
@ -165,16 +165,16 @@ test("hidden control copy never overwrites the visible human or its position", (
expect(merged[0]!.additional_kwargs?.[MESSAGE_SEQ_KEY]).toBe(5);
});
test("tool results dedupe by tool_call_id and keep the known position", () => {
test("copies of one tool result dedupe by message id and keep the known position", () => {
const historyTool = {
id: "tool-old-row",
id: "tool-result",
type: "tool",
content: "partial",
tool_call_id: "call-1",
additional_kwargs: { [MESSAGE_SEQ_KEY]: 6 },
} as Message;
const liveTool = {
id: "tool-new-row",
id: "tool-result",
type: "tool",
content: "complete",
tool_call_id: "call-1",
@ -193,6 +193,55 @@ test("tool results dedupe by tool_call_id and keep the known position", () => {
expect(tool.additional_kwargs?.[MESSAGE_SEQ_KEY]).toBe(6);
});
test.each(["run-a", "run-b"])(
"reused tool-call ids preserve both results and positions (second run %s)",
(secondRunId) => {
const first = msg("tool-first", "tool", "first result", 3, {
tool_call_id: "call_0",
});
const second = msg("tool-second", "tool", "second result", 6, {
tool_call_id: "call_0",
});
const history = buildVisibleHistoryMessages(
[
row("run-a", 2, msg("assistant-first", "ai", "search one")),
row("run-a", 3, first),
row(secondRunId, 5, msg("assistant-second", "ai", "search two")),
row(secondRunId, 6, second),
row(secondRunId, 7, { ...second, content: "complete second result" }),
],
new Set(),
);
const live = [
msg("assistant-second", "ai", "search two"),
msg("tool-second", "tool", "replayed second result", undefined, {
tool_call_id: "call_0",
}),
];
const results = mergeMessages(history, live, []).filter(
(message) => message.type === "tool",
);
expect(idsOf(results)).toEqual(["tool-first", "tool-second"]);
expect(results.map((message) => message.content)).toEqual([
"first result",
"replayed second result",
]);
expect(seqsOf(results)).toEqual([3, 6]);
},
);
test("legacy tool results without message ids are not merged by provider id", () => {
const first = {
type: "tool",
tool_call_id: "call_0",
content: "first",
} as Message;
const second = { ...first, content: "second" } as Message;
expect(messageIdentity(first)).toBeUndefined();
expect(dedupeMessagesByIdentity([first, second])).toEqual([first, second]);
});
test("invalid seq values never overwrite a trusted position or break the merge", () => {
const invalid = [null, "7", Number.NaN, 1.5, Number.MAX_SAFE_INTEGER + 1];
for (const bad of invalid) {

103
frontend/tests/unit/core/threads/tool-result.test.ts
View File

@ -3,6 +3,69 @@ import { expect, test } from "@rstest/core";
import { getToolResultStatus, hasToolResult } from "@/core/threads/hooks";
test.each(["success", "error", "pending"] as const)(
"a unique setup call resolves its leading result as %s",
(status) => {
const messages: Message[] = [
...(status === "pending"
? []
: [
{
type: "tool",
tool_call_id: "call_0",
name: "setup_agent",
content:
status === "error" ? "Error: save failed" : "Agent saved",
status,
} as Message,
]),
{
type: "ai",
id: "setup",
content: "",
tool_calls: [{ id: "call_0", name: "setup_agent", args: {} }],
},
];
expect(getToolResultStatus(messages, "setup_agent")).toBe(status);
},
);
test("a setup result before a human boundary cannot finish the next turn", () => {
const messages = [
{ type: "tool", tool_call_id: "call_0", content: "Agent saved" },
{ type: "human", content: "Try again" },
{
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "setup_agent", args: {} }],
},
] as Message[];
expect(getToolResultStatus(messages, "setup_agent")).toBe("pending");
});
test.each([false, true])(
"an unrelated assistant fences setup status (leading=%s)",
(leading) => {
const owner = {
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "setup_agent", args: {} }],
} as Message;
const other = { type: "ai", content: "An unrelated answer" } as Message;
const result = {
type: "tool",
tool_call_id: "call_0",
content: "Agent saved",
} as Message;
expect(
getToolResultStatus(
leading ? [result, other, owner] : [owner, other, result],
"setup_agent",
),
).toBe("pending");
},
);
test("recognizes a completed tool from its ToolMessage", () => {
const messages = [
{
@ -121,3 +184,43 @@ test("waits for the latest setup attempt instead of reusing an older success", (
expect(getToolResultStatus(messages, "setup_agent")).toBe("pending");
});
test("another tool reusing a provider id cannot replace the setup result", () => {
const messages = [
{
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "setup_agent", args: {} }],
},
{ type: "tool", tool_call_id: "call_0", content: "Saved" },
{
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "web_search", args: {} }],
},
{
type: "tool",
tool_call_id: "call_0",
content: "Error: failed",
status: "error",
},
] as Message[];
expect(getToolResultStatus(messages, "setup_agent")).toBe("success");
});
test("a reused provider id cannot complete a new pending setup attempt", () => {
const messages = [
{
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "setup_agent", args: {} }],
},
{ type: "tool", tool_call_id: "call_0", content: "Saved" },
{
type: "ai",
content: "",
tool_calls: [{ id: "call_0", name: "setup_agent", args: {} }],
},
] as Message[];
expect(getToolResultStatus(messages, "setup_agent")).toBe("pending");
});