mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
fix(code): ensure unique message widget IDs on history load (#4454)
Fix intermittent "Could not load history" errors caused by duplicate message widget IDs when resuming or switching threads. --- Resume/thread-switch loads were intermittently failing with `Could not load history: Tried to insert a widget with ID 'msg-XXXX', but a widget already exists...`. Message IDs were only 32 bits (`uuid4().hex[:8]`), so IDs could collide, and `_load_thread_history` had no defense against mounting a widget whose ID was already in the DOM (e.g. a re-entrant load over surviving widgets). This widens auto-generated message/assistant IDs to the full 128-bit uuid4 hex and makes the history mount skip any widget whose ID is already present, logging a warning instead of aborting the load. Made by [Open SWE](https://openswe.vercel.app/agents/e6466f87-1dba-5949-34eb-29774cd3eaa6) --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
@@ -10795,32 +10795,86 @@ class DeepAgentsApp(App):
|
||||
if payload.context_tokens > 0:
|
||||
self._on_tokens_update(payload.context_tokens)
|
||||
|
||||
# 3. Bulk load into store (sets visible window)
|
||||
_archived, visible = self._message_store.bulk_load(payload.messages)
|
||||
|
||||
# 5. Cache container ref (single query)
|
||||
# 5. Cache container ref (single query). Queried before the store
|
||||
# load so history can be reconciled against widgets already in the
|
||||
# DOM (see below).
|
||||
try:
|
||||
messages_container = self.query_one("#messages", Container)
|
||||
except NoMatches:
|
||||
return
|
||||
|
||||
# 6-7. Create and mount only visible widgets (max WINDOW_SIZE)
|
||||
widgets = [msg_data.to_widget() for msg_data in visible]
|
||||
if widgets:
|
||||
nodes: list[Widget] = []
|
||||
for widget, msg_data in zip(widgets, visible, strict=False):
|
||||
nodes.append(widget)
|
||||
footer = self._build_message_timestamp_footer(
|
||||
msg_data, visible=self._message_timestamps_visible
|
||||
# 3. Reconcile against existing state before loading the store.
|
||||
# Mounting a widget whose ID already exists raises `DuplicateIds`,
|
||||
# which would abort the entire history load. Widened message IDs
|
||||
# make natural collisions vanishingly unlikely, but a re-entrant
|
||||
# load (e.g. a server respawn that re-runs the startup sequence
|
||||
# over a non-cleared store and its surviving widgets) can still
|
||||
# reintroduce an already-present ID. Two guards keep this safe:
|
||||
#
|
||||
# a) Drop payload messages whose ID is already in the store (or
|
||||
# repeated within the payload) before `bulk_load`. Otherwise
|
||||
# `bulk_load` would append duplicate entries to `_messages`,
|
||||
# desyncing the visible window from the DOM and tripping up
|
||||
# later pruning/hydration.
|
||||
# b) Skip mounting any visible message whose ID is already in the
|
||||
# DOM. `bulk_load` returns a window over the *whole* store, so
|
||||
# a surviving pre-existing entry can still surface as a mount
|
||||
# candidate even after (a); its widget already exists.
|
||||
seen: set[str] = set()
|
||||
deduped: list[MessageData] = []
|
||||
for msg_data in payload.messages:
|
||||
if (
|
||||
msg_data.id in seen
|
||||
or self._message_store.get_message(msg_data.id) is not None
|
||||
):
|
||||
continue
|
||||
seen.add(msg_data.id)
|
||||
deduped.append(msg_data)
|
||||
dropped = len(payload.messages) - len(deduped)
|
||||
if dropped:
|
||||
logger.warning(
|
||||
"Dropped %d duplicate history message(s) for thread %s: "
|
||||
"IDs were already in the store or repeated in the payload",
|
||||
dropped,
|
||||
history_thread_id,
|
||||
)
|
||||
|
||||
# Bulk load into store (sets visible window over the deduped set).
|
||||
_archived, visible = self._message_store.bulk_load(deduped)
|
||||
|
||||
# 6-7. Create and mount the visible widgets (max WINDOW_SIZE),
|
||||
# skipping any whose ID is already mounted (guard (b) above).
|
||||
# `existing_ids` includes footer node IDs, which never collide with
|
||||
# the `msg-`/`asst-` message IDs checked here.
|
||||
existing_ids = {
|
||||
node.id for node in messages_container.children if node.id is not None
|
||||
}
|
||||
mounted: list[tuple[Widget, MessageData]] = []
|
||||
nodes: list[Widget] = []
|
||||
for msg_data in visible:
|
||||
if msg_data.id in existing_ids:
|
||||
logger.debug(
|
||||
"Skipping already-mounted history widget %s in thread %s",
|
||||
msg_data.id,
|
||||
history_thread_id,
|
||||
)
|
||||
if footer is not None:
|
||||
nodes.append(footer)
|
||||
continue
|
||||
existing_ids.add(msg_data.id)
|
||||
widget = msg_data.to_widget()
|
||||
mounted.append((widget, msg_data))
|
||||
nodes.append(widget)
|
||||
footer = self._build_message_timestamp_footer(
|
||||
msg_data, visible=self._message_timestamps_visible
|
||||
)
|
||||
if footer is not None:
|
||||
nodes.append(footer)
|
||||
if nodes:
|
||||
await messages_container.mount(*nodes)
|
||||
|
||||
# 8. Render content for AssistantMessage after mount
|
||||
assistant_updates = [
|
||||
widget.set_content(msg_data.content)
|
||||
for widget, msg_data in zip(widgets, visible, strict=False)
|
||||
for widget, msg_data in mounted
|
||||
if isinstance(widget, AssistantMessage) and msg_data.content
|
||||
]
|
||||
if assistant_updates:
|
||||
|
||||
@@ -1025,7 +1025,7 @@ async def execute_task_textual(
|
||||
# Get or create assistant message for this namespace
|
||||
current_msg = assistant_message_by_namespace.get(ns_key)
|
||||
if current_msg is None:
|
||||
msg_id = f"asst-{uuid.uuid4().hex[:8]}"
|
||||
msg_id = f"asst-{uuid.uuid4().hex}"
|
||||
# Mark active BEFORE mounting so pruning
|
||||
# (triggered by mount) won't remove it
|
||||
# (_mount_message can trigger
|
||||
@@ -1828,7 +1828,7 @@ async def _flush_assistant_text_ns(
|
||||
current_msg = assistant_message_by_namespace.get(ns_key)
|
||||
if current_msg is None:
|
||||
# No message was created during streaming - create one with full content
|
||||
msg_id = f"asst-{uuid.uuid4().hex[:8]}"
|
||||
msg_id = f"asst-{uuid.uuid4().hex}"
|
||||
current_msg = AssistantMessage(text, id=msg_id)
|
||||
await adapter._mount_message(current_msg)
|
||||
await current_msg.write_initial_content()
|
||||
|
||||
@@ -111,8 +111,13 @@ class MessageData:
|
||||
`tool_args` instead.
|
||||
"""
|
||||
|
||||
id: str = field(default_factory=lambda: f"msg-{uuid.uuid4().hex[:8]}")
|
||||
"""Unique identifier used to match the dataclass to its DOM widget."""
|
||||
id: str = field(default_factory=lambda: f"msg-{uuid.uuid4().hex}")
|
||||
"""Unique identifier used to match the dataclass to its DOM widget.
|
||||
|
||||
Uses the full 128-bit `uuid4` hex (not a truncated prefix) so IDs stay
|
||||
unique across large histories and long sessions; a widget-ID collision
|
||||
raises `DuplicateIds` when the widget is mounted.
|
||||
"""
|
||||
|
||||
timestamp: float = field(default_factory=time)
|
||||
"""Unix epoch timestamp of when the message was created."""
|
||||
@@ -298,7 +303,7 @@ class MessageData:
|
||||
UserMessage,
|
||||
)
|
||||
|
||||
widget_id = widget.id or f"msg-{uuid.uuid4().hex[:8]}"
|
||||
widget_id = widget.id or f"msg-{uuid.uuid4().hex}"
|
||||
|
||||
if isinstance(widget, SkillMessage):
|
||||
return cls(
|
||||
|
||||
@@ -8027,6 +8027,211 @@ class TestMessageTimestampFooters:
|
||||
with pytest.raises(NoMatches):
|
||||
app.query_one("#hist-app-timestamp-footer", Static)
|
||||
|
||||
async def test_load_thread_history_skips_duplicate_ids(self) -> None:
|
||||
"""History reusing an already-mounted widget ID is skipped, not fatal.
|
||||
|
||||
Regression: mounting a widget whose ID already exists raises
|
||||
`DuplicateIds`, which previously aborted the whole load and surfaced a
|
||||
"Could not load history" note instead of the conversation.
|
||||
"""
|
||||
from deepagents_code.app import _ThreadHistoryPayload
|
||||
from deepagents_code.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
# Pre-mount a widget occupying the ID the history will reuse.
|
||||
await app._mount_message(AppMessage("stale", id="dup-id"))
|
||||
await pilot.pause()
|
||||
|
||||
payload = _ThreadHistoryPayload(
|
||||
[
|
||||
MessageData(type=MessageType.USER, content="dup", id="dup-id"),
|
||||
MessageData(type=MessageType.USER, content="fresh", id="fresh-id"),
|
||||
],
|
||||
0,
|
||||
"",
|
||||
)
|
||||
await app._load_thread_history(thread_id="t-dup", preloaded_payload=payload)
|
||||
await pilot.pause()
|
||||
|
||||
# The load completed without the fatal "Could not load history" note.
|
||||
notes = [str(widget._content) for widget in app.query(AppMessage)]
|
||||
assert not any("Could not load history" in note for note in notes)
|
||||
# The colliding message was skipped; the original widget survives
|
||||
# (exactly one, no duplicate mounted) and keeps its own content --
|
||||
# the history entry did not overwrite or replace it.
|
||||
survivors = app.query("#dup-id")
|
||||
assert len(survivors) == 1
|
||||
survivor = survivors.first()
|
||||
assert isinstance(survivor, AppMessage)
|
||||
assert "stale" in str(survivor._content)
|
||||
# The non-colliding message mounted normally.
|
||||
assert app.query_one("#fresh-id", UserMessage)
|
||||
# The load ran to completion (past the mount block to step 9).
|
||||
assert any("Resumed thread: t-dup" in note for note in notes)
|
||||
|
||||
async def test_load_thread_history_preserves_assistant_content_after_skip(
|
||||
self,
|
||||
) -> None:
|
||||
"""A skipped duplicate keeps the surviving assistant content aligned.
|
||||
|
||||
Regression guard for the `mounted` pairing: `set_content` must render
|
||||
each surviving `AssistantMessage`'s own content, never a neighbor's.
|
||||
Only exercised when a skip removes an entry, shifting the survivors.
|
||||
"""
|
||||
from deepagents_code.app import _ThreadHistoryPayload
|
||||
from deepagents_code.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
# Occupy the first assistant message's ID so it is skipped.
|
||||
await app._mount_message(AssistantMessage("stale", id="asst-dup"))
|
||||
await pilot.pause()
|
||||
|
||||
payload = _ThreadHistoryPayload(
|
||||
[
|
||||
MessageData(type=MessageType.ASSISTANT, content="A", id="asst-dup"),
|
||||
MessageData(
|
||||
type=MessageType.ASSISTANT, content="B", id="asst-fresh"
|
||||
),
|
||||
],
|
||||
0,
|
||||
"",
|
||||
)
|
||||
await app._load_thread_history(
|
||||
thread_id="t-asst", preloaded_payload=payload
|
||||
)
|
||||
await pilot.pause()
|
||||
|
||||
# The surviving fresh assistant renders its own content ("B"),
|
||||
# not the skipped duplicate's ("A").
|
||||
fresh = app.query_one("#asst-fresh", AssistantMessage)
|
||||
assert "B" in str(fresh._content)
|
||||
assert "A" not in str(fresh._content)
|
||||
|
||||
async def test_load_thread_history_dedupes_within_payload(self) -> None:
|
||||
"""Two payload entries sharing an ID mount exactly one widget.
|
||||
|
||||
The intra-batch `seen` guard prevents a same-payload collision from
|
||||
raising `DuplicateIds` on the bulk mount.
|
||||
"""
|
||||
from deepagents_code.app import _ThreadHistoryPayload
|
||||
from deepagents_code.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
payload = _ThreadHistoryPayload(
|
||||
[
|
||||
MessageData(type=MessageType.USER, content="first", id="same-id"),
|
||||
MessageData(type=MessageType.USER, content="second", id="same-id"),
|
||||
MessageData(type=MessageType.USER, content="other", id="other-id"),
|
||||
],
|
||||
0,
|
||||
"",
|
||||
)
|
||||
await app._load_thread_history(
|
||||
thread_id="t-intra", preloaded_payload=payload
|
||||
)
|
||||
await pilot.pause()
|
||||
|
||||
notes = [str(widget._content) for widget in app.query(AppMessage)]
|
||||
assert not any("Could not load history" in note for note in notes)
|
||||
# Exactly one widget for the repeated ID; the first entry wins.
|
||||
assert len(app.query("#same-id")) == 1
|
||||
same = app.query_one("#same-id", UserMessage)
|
||||
assert "first" in str(same._content)
|
||||
assert app.query_one("#other-id", UserMessage)
|
||||
|
||||
async def test_load_thread_history_keeps_store_window_in_sync_after_skip(
|
||||
self,
|
||||
) -> None:
|
||||
"""A re-entrant load must not create duplicate store entries.
|
||||
|
||||
Regression: `bulk_load` blindly appends, so re-loading a message whose
|
||||
ID is already in the store used to add a second `_messages` entry for
|
||||
the same ID -- desyncing the visible window from the DOM and tripping
|
||||
up later pruning/hydration. Deduplicating against the store before
|
||||
`bulk_load` keeps every ID represented exactly once.
|
||||
"""
|
||||
from deepagents_code.app import _ThreadHistoryPayload
|
||||
from deepagents_code.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
# `_mount_message` records the widget in the store, so `dup-id` is
|
||||
# present in both the store and the DOM before the load.
|
||||
await app._mount_message(AppMessage("stale", id="dup-id"))
|
||||
await pilot.pause()
|
||||
|
||||
payload = _ThreadHistoryPayload(
|
||||
[
|
||||
MessageData(type=MessageType.USER, content="dup", id="dup-id"),
|
||||
MessageData(type=MessageType.USER, content="fresh", id="fresh-id"),
|
||||
],
|
||||
0,
|
||||
"",
|
||||
)
|
||||
await app._load_thread_history(
|
||||
thread_id="t-sync", preloaded_payload=payload
|
||||
)
|
||||
await pilot.pause()
|
||||
|
||||
store = app._message_store
|
||||
# Every stored ID is represented exactly once (no phantom double
|
||||
# entry for the re-loaded `dup-id`).
|
||||
all_ids = [msg.id for msg in store.get_all_messages()]
|
||||
assert len(all_ids) == len(set(all_ids))
|
||||
assert store.get_message("dup-id") is not None
|
||||
assert store.get_message("fresh-id") is not None
|
||||
# The visible window is internally consistent with its range.
|
||||
start, end = store.get_visible_range()
|
||||
assert store.visible_count == end - start
|
||||
|
||||
async def test_load_thread_history_all_duplicates_completes_cleanly(self) -> None:
|
||||
"""A payload whose every ID is already mounted still finishes the load.
|
||||
|
||||
With nothing left to mount after dedup, the load must skip the mount
|
||||
block and still reach completion without an error note -- the benign
|
||||
re-entrant reload case (same payload over surviving widgets).
|
||||
"""
|
||||
from deepagents_code.app import _ThreadHistoryPayload
|
||||
from deepagents_code.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
await app._mount_message(UserMessage("kept", id="a-id"))
|
||||
await app._mount_message(UserMessage("kept", id="b-id"))
|
||||
await pilot.pause()
|
||||
|
||||
payload = _ThreadHistoryPayload(
|
||||
[
|
||||
MessageData(type=MessageType.USER, content="a", id="a-id"),
|
||||
MessageData(type=MessageType.USER, content="b", id="b-id"),
|
||||
],
|
||||
0,
|
||||
"",
|
||||
)
|
||||
await app._load_thread_history(thread_id="t-all", preloaded_payload=payload)
|
||||
await pilot.pause()
|
||||
|
||||
notes = [str(widget._content) for widget in app.query(AppMessage)]
|
||||
assert not any("Could not load history" in note for note in notes)
|
||||
assert any("Resumed thread: t-all" in note for note in notes)
|
||||
# No duplicates were mounted; the originals survive.
|
||||
assert len(app.query("#a-id")) == 1
|
||||
assert len(app.query("#b-id")) == 1
|
||||
|
||||
async def test_footers_render_for_hydrated_messages_above(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
|
||||
@@ -216,6 +216,41 @@ class TestMessageData:
|
||||
assert data.is_streaming is False
|
||||
assert data.height_hint is None
|
||||
|
||||
def test_default_ids_use_full_uuid_hex(self):
|
||||
"""Auto-generated IDs use the full 128-bit hex, not a truncated prefix.
|
||||
|
||||
A wider ID keeps widget IDs unique across large histories and long
|
||||
sessions; a collision raises `DuplicateIds` when the widget mounts.
|
||||
"""
|
||||
ids = {
|
||||
MessageData(type=MessageType.USER, content="test").id for _ in range(1000)
|
||||
}
|
||||
# 1000 distinct IDs, each a full uuid4 hex suffix.
|
||||
assert len(ids) == 1000
|
||||
for message_id in ids:
|
||||
assert message_id.startswith("msg-")
|
||||
suffix = message_id.removeprefix("msg-")
|
||||
assert len(suffix) == 32
|
||||
# A full uuid4 hex suffix is valid hexadecimal.
|
||||
int(suffix, 16)
|
||||
|
||||
def test_from_widget_fallback_id_uses_full_uuid_hex(self):
|
||||
"""`from_widget` synthesizes a full-hex ID when the widget has none.
|
||||
|
||||
A widget mounted without an explicit ID must still get a collision-safe
|
||||
128-bit identifier, matching the `MessageData` default.
|
||||
"""
|
||||
# Constructed without an `id` kwarg, so `widget.id` is None and the
|
||||
# fallback in `from_widget` must synthesize one.
|
||||
widget = UserMessage("no id")
|
||||
|
||||
data = MessageData.from_widget(widget)
|
||||
|
||||
assert data.id.startswith("msg-")
|
||||
suffix = data.id.removeprefix("msg-")
|
||||
assert len(suffix) == 32
|
||||
int(suffix, 16)
|
||||
|
||||
def test_tool_message_requires_tool_name(self):
|
||||
"""Test that TOOL messages must have a tool_name."""
|
||||
with pytest.raises(ValueError, match="TOOL messages must have a tool_name"):
|
||||
|
||||
Reference in New Issue
Block a user