Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 53 additions & 1 deletion libs/code/deepagents_code/_textual_patches.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
r"""Runtime patches over Textual internals, imported for side effect.

This module hosts three independent best-effort patches over private Textual
This module hosts four independent best-effort patches over private Textual
APIs. Each guards its own import/assignment and degrades to stock Textual
behavior (logging a warning) if the targeted internals move, so they have
separate lifecycles — do not delete the whole file when only one lands
Expand Down Expand Up @@ -42,6 +42,15 @@
drag) to word boundaries. No upstream issue tracks this yet, so it has
no removal criterion — it stays until Textual grows native word select.

4. Detached-widget hit filtering. The compositor keeps reporting a widget as
visible for a few event-loop iterations after it leaves the DOM, which
`Markdown.update` (and therefore the `MarkdownStream` that drives every
streaming assistant message) does constantly. `Screen._forward_event`
starts a text selection from `content_widget.parent`, which is `None` for
such a widget, so a mouse press landing on freshly replaced markdown
crashes the app with `AttributeError: 'NoneType' object has no attribute
'region'`. Tracked in Textualize/textual#6643; remove when that lands.

Imported for side effect from `app.py` before any `App()` is created.
"""

Expand Down Expand Up @@ -420,3 +429,46 @@ async def _on_click_with_word_select(self: Widget, event: Click) -> None:
_textual_version,
exc,
)


try:
from textual.screen import Screen as _HitScreen

_original_get_widget_and_offset_at = _HitScreen.get_widget_and_offset_at
except (ImportError, AttributeError) as exc: # pragma: no cover - defensive
logger.warning(
"Textual detached-hit patch skipped (textual %s): %s",
_textual_version,
exc,
)
else:

def _get_widget_and_offset_at_attached(
self: Screen,
x: int,
y: int,
) -> tuple[Widget | None, Offset | None]:
"""Ignore compositor hits on widgets that already left the DOM.

Returns:
The stock result, or `(None, None)` when the hit widget is
detached, which sends Textual down its existing "nothing
selectable here" branch instead of dereferencing a `None` parent.
"""
widget, offset = _original_get_widget_and_offset_at(self, x, y)
if (
widget is not None
and not isinstance(widget, _HitScreen)
and (widget.parent is None or not widget.is_attached)
):
return None, None
return widget, offset

try:
_HitScreen.get_widget_and_offset_at = _get_widget_and_offset_at_attached # ty: ignore[invalid-assignment]
except (AttributeError, TypeError) as exc: # pragma: no cover - defensive
logger.warning(
"Textual detached-hit patch assignment rejected (textual %s): %s",
_textual_version,
exc,
)
48 changes: 48 additions & 0 deletions libs/code/tests/unit_tests/test_textual_patches.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from pathlib import Path

import pytest
from textual import events
from textual._time import get_time
from textual._xterm_parser import XTermParser
from textual.app import App, ComposeResult
Expand Down Expand Up @@ -77,6 +78,53 @@ async def test_triple_click_selects_clicked_widget_not_history(self) -> None:
assert pilot.app.screen.get_selected_text() == "second message"


class TestDetachedHitGuard:
"""Coverage of the Textualize/textual#6643 crash guard."""

async def test_mouse_down_on_detached_widget_does_not_crash(self) -> None:
"""A press on a widget pruned since the last repaint must be ignored.

`Markdown.update` — which `MarkdownStream` runs on every streaming
assistant message — detaches its old blocks while the compositor still
reports them as visible. `_detach` is exactly what Textual calls during
that prune, so calling it directly pins the race window deterministically
instead of spinning the event loop until it happens to be observed.
Without the guard, `Screen._forward_event` raises `AttributeError` on the
detached widget's `None` parent and takes the whole app down.
"""
async with SelectableMarkdownApp().run_test() as pilot:
screen = pilot.app.screen
document = pilot.app.query_one("#msg", Markdown)
paragraph = document.query("*").first()
x = paragraph.region.x + 1
y = paragraph.region.y
assert screen._compositor.get_widget_and_offset_at(x, y)[0] is paragraph

paragraph._detach()
try:
assert screen.get_widget_and_offset_at(x, y) == (None, None)
screen._forward_event(
events.MouseDown(None, x, y, 0, 0, 1, False, False, False)
)

assert screen._select_state is None
finally:
# Textual's own teardown asserts every widget still has a
# parent, so hand the simulated prune victim back to the DOM.
paragraph._attach(document)

async def test_attached_widget_hit_is_still_reported(self) -> None:
"""The guard must only drop detached hits, not live ones."""
async with SelectableTextApp().run_test() as pilot:
widget = pilot.app.query_one("#msg", Static)
offset = widget.content_region.offset + Offset(2, 0)

hit, hit_offset = pilot.app.screen.get_widget_and_offset_at(*offset)

assert hit is widget
assert hit_offset == Offset(2, 0)


class TestPatchedSequenceToKeyEvents:
r"""Targeted coverage of the two interventions in the shim."""

Expand Down