fix(code): serialize transcript tail reconciliation (#6143)

Long transcripts no longer duplicate rows when new output arrives during
history hydration.

---

The bounded tail jump introduced by #6057 could overlap with
scroll-triggered hydration. Both paths built widgets from the same stale
visible range, so the second mount hit duplicate DOM IDs and could drop
fresh output or desynchronize the transcript store.

Serialize transcript store/DOM mutations across append, hydration,
pruning, and clear operations. The tail jump now derives mounted IDs
from the actual container and releases removed tool-group summaries
before regrouping surviving rows.

Made by [Open
SWE](https://openswe.vercel.app/agents/708f22e9-c9ed-554d-858f-1c2090a9482b)

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

Author: mdrxy

libs/code/deepagents_code/app.py

@@ -4302,6 +4302,9 @@ def __init__(
         self._transcript_generation = 0
         """Invalidates hydration/pruning work when the transcript is cleared."""
 
+        self._transcript_mutation_lock = asyncio.Lock()
+        """Serializes message-store and transcript DOM reconciliation."""
+
         self._hydration_requests: set[Literal["above", "below"]] = set()
         """Coalesced transcript hydration directions awaiting one UI slice."""
 
@@ -9390,9 +9393,19 @@ async def _hydrate_messages(
     ) -> int:
         """Hydrate one contiguous batch at a mounted-window edge.
 
-        Args:
-            direction: Edge receiving stored messages.
-            count: Maximum messages to mount; defaults to `HYDRATE_BUFFER`.
+        Returns:
+            Number of messages mounted.
+        """
+        async with self._transcript_mutation_lock:
+            return await self._hydrate_messages_unlocked(direction, count=count)
+
+    async def _hydrate_messages_unlocked(
+        self,
+        direction: Literal["above", "below"],
+        *,
+        count: int | None = None,
+    ) -> int:
+        """Hydrate one batch while transcript mutation is already serialized.
 
         Returns:
             Number of messages mounted.
@@ -19984,6 +19997,22 @@ async def _mount_message(
         | ReasoningMessage
         | ToolCallMessage
         | SkillMessage,
+    ) -> bool:
+        """Mount one message while transcript mutation is serialized.
+
+        Returns:
+            Whether the widget reached the screen.
+        """
+        async with self._transcript_mutation_lock:
+            return await self._mount_message_unlocked(widget)
+
+    async def _mount_message_unlocked(
+        self,
+        widget: Static
+        | AssistantMessage
+        | ReasoningMessage
+        | ToolCallMessage
+        | SkillMessage,
     ) -> bool:
         """Mount a message widget to the messages area.
 
@@ -20118,13 +20147,18 @@ async def _move_transcript_window_to_tail(
         if tail is None:
             while self._message_store.has_messages_below:
                 before = self._message_store.get_visible_range()[1]
-                await self._hydrate_messages("below")
+                await self._hydrate_messages_unlocked("below")
                 if self._message_store.get_visible_range()[1] == before:
                     return False
             return True
 
         generation = self._transcript_generation
-        mounted_ids = {data.id for data in self._message_store.get_visible_messages()}
+        mounted_ids = {
+            child.id
+            for child in messages_container.children
+            if child.id is not None
+            and self._message_store.get_message(child.id) is not None
+        }
         tail_ids = {data.id for data in tail}
         entries = [
             self._build_hydration_entry(data)
@@ -20163,6 +20197,9 @@ async def _move_transcript_window_to_tail(
             ]
             await messages_container.remove_children(hydrated_nodes)
             return False
+        for child in obsolete:
+            if isinstance(child, ToolGroupSummary):
+                child._release_all_collapsible()
         await messages_container.remove_children(obsolete)
         self._schedule_message_height_measurements([data.id for data in moved])
         self._sync_transcript_spacers(messages_container)
@@ -20176,12 +20213,24 @@ async def _prune_messages(
         *,
         count: int | None = None,
     ) -> int:
-        """Prune a bounded batch from one mounted-window edge.
+        """Prune one mounted-window edge while mutation is serialized.
 
-        Args:
-            direction: Edge to remove messages from.
-            messages_container: Cached transcript container, when available.
-            count: Maximum messages to remove; defaults to the soft-limit excess.
+        Returns:
+            Number of messages removed.
+        """
+        async with self._transcript_mutation_lock:
+            return await self._prune_messages_unlocked(
+                direction, messages_container, count=count
+            )
+
+    async def _prune_messages_unlocked(
+        self,
+        direction: Literal["above", "below"],
+        messages_container: Container | None = None,
+        *,
+        count: int | None = None,
+    ) -> int:
+        """Prune a bounded batch while transcript mutation is serialized.
 
         Returns:
             Number of messages removed.
@@ -20535,6 +20584,11 @@ def on_user_message_expansion_changed(
             self._schedule_message_height_measurements([event.widget.id])
 
     async def _clear_messages(self) -> None:
+        """Clear the transcript while mutation is serialized."""
+        async with self._transcript_mutation_lock:
+            await self._clear_messages_unlocked()
+
+    async def _clear_messages_unlocked(self) -> None:
         """Clear the messages area and message store."""
         self._transcript_generation += 1
         # Drop buffered `!` shell output so it never leaks across a thread

libs/code/tests/unit_tests/test_app.py

@@ -11271,6 +11271,87 @@ async def count_hydration(
                 with pytest.raises(NoMatches):
                     app.query_one(f"#{message_id}", UserMessage)
 
+    async def test_mount_message_waits_for_inflight_hydration(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Concurrent hydration and append keep one mounted row per store entry."""
+        from deepagents_code.tui.widgets.message_store import MessageData, MessageType
+
+        app = DeepAgentsApp()
+        monkeypatch.setattr(app._message_store, "WINDOW_SIZE", 3)
+        monkeypatch.setattr(app, "_schedule_transcript_prune", lambda *_args: None)
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            history = [
+                MessageData(
+                    type=MessageType.USER,
+                    content=f"m{index}",
+                    id=f"race-{index}",
+                )
+                for index in range(6)
+            ]
+            app._message_store.bulk_load(history)
+            messages = app.query_one("#messages", Container)
+            tail_entries = [
+                app._build_hydration_entry(data)
+                for data in app._message_store.get_visible_messages()
+            ]
+            assert await app._mount_hydration_batch(
+                messages,
+                tail_entries,
+                generation=app._transcript_generation,
+            )
+            await messages.remove_children(
+                [
+                    child
+                    for child in messages.children
+                    if child.id
+                    and any(
+                        child.id.startswith(f"race-{index}") for index in range(3, 6)
+                    )
+                ]
+            )
+            app._message_store._visible_start = 0
+            app._message_store._visible_end = 2
+            head_entries = [
+                app._build_hydration_entry(data)
+                for data in app._message_store.get_visible_messages()
+            ]
+            assert await app._mount_hydration_batch(
+                messages,
+                head_entries,
+                generation=app._transcript_generation,
+            )
+
+            hydration_started = asyncio.Event()
+            release_hydration = asyncio.Event()
+            mount_batch = app._mount_hydration_batch
+
+            async def pause_first_batch(*args: Any, **kwargs: Any) -> bool:
+                if not hydration_started.is_set():
+                    hydration_started.set()
+                    await release_hydration.wait()
+                return await mount_batch(*args, **kwargs)
+
+            monkeypatch.setattr(app, "_mount_hydration_batch", pause_first_batch)
+            hydrate = asyncio.create_task(app._hydrate_messages("below", count=3))
+            await hydration_started.wait()
+            append = asyncio.create_task(
+                app._mount_message(UserMessage("new", id="race-new"))
+            )
+            await asyncio.sleep(0)
+            assert not append.done()
+
+            release_hydration.set()
+            assert await hydrate == 3
+            assert await append is True
+            await pilot.pause()
+
+            assert app._message_store.get_visible_range() == (3, 7)
+            for message_id in ["race-3", "race-4", "race-5", "race-new"]:
+                assert len(app.query(f"#{message_id}")) == 1
+
     async def test_mount_message_hydrates_tail_blocked_by_protected_row(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None: