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
10 changes: 3 additions & 7 deletions libs/code/deepagents_code/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -20122,12 +20122,14 @@ def _on_subagent_event(self, event: dict[str, Any]) -> None:
panel.on_subagent_event(event)

async def _on_auto_mode_event(self, event: dict[str, Any]) -> None:
"""Render one compact sanitized Auto event in the transcript.
"""Render one compact sanitized Auto control-state notice.

Args:
event: Validated custom-stream event from the server middleware.
"""
kind = event.get("event")
if kind not in {"fallback", "warning"}:
return
reason = str(event.get("reason") or "")
if kind == "fallback":
if event.get("mode") == "manual":
Expand All @@ -20146,12 +20148,6 @@ async def _on_auto_mode_event(self, event: dict[str, Any]) -> None:
f"unavailable {event.get('consecutive_unavailable', 0)}, "
f"total {event.get('total_denials', 0)})."
)
elif kind == "denial":
text = f"Auto denied [{event.get('category', 'policy')}]: {reason}"
elif kind == "unavailable":
# Reason is a short cause fragment from the server; keep the UI
# line outcome-focused so fail-closed denial is obvious.
text = f"Auto classifier unavailable: {reason} — tool not executed"
else:
text = f"Auto warning: {reason}"
await self._mount_message(AppMessage(text))
Expand Down
4 changes: 2 additions & 2 deletions libs/code/deepagents_code/tui/textual_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -1092,7 +1092,7 @@ def _require_approval_mode_key(value: str | None) -> str:


def _is_renderable_auto_mode_event(data: Any, *, is_main_agent: bool) -> bool: # noqa: ANN401
"""Return whether a custom event is a sanitized top-level Auto event."""
"""Return whether a custom event is a sanitized Auto control-state notice."""
if (
not is_main_agent
or not isinstance(data, dict)
Expand All @@ -1103,7 +1103,7 @@ def _is_renderable_auto_mode_event(data: Any, *, is_main_agent: bool) -> bool:
reason = data.get("reason")
mode = data.get("mode")
return (
event in {"denial", "unavailable", "fallback", "warning"}
event in {"fallback", "warning"}
and (reason is None or isinstance(reason, str))
and (mode is None or (event == "fallback" and mode == "manual"))
)
Expand Down
10 changes: 10 additions & 0 deletions libs/code/tests/unit_tests/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -30369,6 +30369,16 @@ async def test_server_manual_fallback_updates_tui_mode_and_warns(self) -> None:
)
mount.assert_awaited_once()

@pytest.mark.parametrize("kind", ["denial", "unavailable"])
async def test_tool_outcome_auto_event_is_not_mounted(self, kind: str) -> None:
app = DeepAgentsApp()
event = {"event": kind, "reason": "tool was denied"}

with patch.object(app, "_mount_message", new=AsyncMock()) as mount:
await app._on_auto_mode_event(event)

mount.assert_not_awaited()


class TestExternalBypassFieldHonored:
"""`event.bypass` overrides queue when set on a prompt event."""
Expand Down
17 changes: 17 additions & 0 deletions libs/code/tests/unit_tests/tui/test_textual_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@
_handle_interrupt_cleanup,
_interrupt_owned_tool_rows,
_is_auto_mode_classifier_chunk,
_is_renderable_auto_mode_event,
_is_summarization_chunk,
_read_mentioned_file,
_session_cost_pricing_ok,
Expand Down Expand Up @@ -1517,6 +1518,22 @@ def test_returns_false_for_unrelated_metadata(self) -> None:
assert _is_auto_mode_classifier_chunk({"lc_source": "summarization"}) is False


class TestIsRenderableAutoModeEvent:
"""Tests for standalone Auto control-state notice filtering."""

@pytest.mark.parametrize("event", ["fallback", "warning"])
def test_accepts_control_state_notice(self, event: str) -> None:
payload = {"type": "auto_mode", "event": event, "reason": "state changed"}

assert _is_renderable_auto_mode_event(payload, is_main_agent=True) is True

@pytest.mark.parametrize("event", ["denial", "unavailable"])
def test_rejects_tool_outcome_event(self, event: str) -> None:
payload = {"type": "auto_mode", "event": event, "reason": "tool was denied"}

assert _is_renderable_auto_mode_event(payload, is_main_agent=True) is False


class TestFormatRubricEvent:
"""Tests for rubric custom-stream event formatting."""

Expand Down
Loading