langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-22 01:35:28 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

feat(code,quickjs): dynamic subagents UI (#4221)

Browse Source 
### Summary

This PR implements a `dcode` ui for dynamic subagents (see attached
images). The implementation on the quickjs side adds code to emit
subagent events into a stream writer using the `custom` channel. `dcode`
then reads these events from the stream and is able to reliably classify
subagent events using the `subagent` stream event type.

### Tests
Unit tests and e2e testing with `dcode` to verify and validate quality +
behavior.

### Dynamic Subagents UI
Collapsed:
<img width="1728" height="1024" alt="Screenshot 2026-06-24 at 7 52
56 AM"
src="https://github.com/user-attachments/assets/34cece20-e072-40e2-a46d-20567583f252"
/>

Expanded:
<img width="1728" height="1016" alt="Screenshot 2026-06-24 at 7 48
42 AM"
src="https://github.com/user-attachments/assets/01e83427-aeee-44f0-b29c-d9f1792db76f"
/>

---------

Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <github@mdrxy.com>
Co-authored-by: Mason Daugherty <mason@langchain.dev>
This commit is contained in:
Colin Francis
2026-06-24 23:57:24 -07:00
committed by GitHub
parent cf536f3399
commit 10bcba2560
21 changed files with 2303 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/code/deepagents_code/app.py
+68
View File
@@ -91,6 +91,7 @@ from deepagents_code.widgets.messages import (
UserMessage,
)
from deepagents_code.widgets.status import StatusBar
from deepagents_code.widgets.subagent_panel import SubagentPanel
from deepagents_code.widgets.welcome import WelcomeBanner
logger = logging.getLogger(__name__)
@@ -970,6 +971,20 @@ def _format_model_params(extra_kwargs: dict[str, Any] | None) -> str:
return f" with model params {json.dumps(extra_kwargs, sort_keys=True)}"
def _display_model_label(spec: str | None) -> str | None:
"""Strip the provider prefix from a model spec for display.
`anthropic:opus` becomes `opus`; only the first colon splits, so a model
name that itself contains a colon is preserved. A spec without a colon (or
an empty/`None` spec) is returned unchanged. This is a cosmetic label only,
so a malformed spec degrades to a slightly-off label rather than an error.
Returns:
The display label, or the spec unchanged when there is no prefix.
"""
return spec.split(":", 1)[1] if spec and ":" in spec else spec
InputMode = Literal["normal", "shell", "shell_incognito", "command"]
_RECONNECT_FORCE_TOKENS: frozenset[str] = frozenset({"force", "--force", "-f"})
@@ -1480,6 +1495,7 @@ class DeepAgentsApp(App):
),
Binding("ctrl+d", "quit_app", "Quit", show=False, priority=True),
Binding("ctrl+t", "toggle_auto_approve", "Toggle Auto-Approve", show=False),
Binding("ctrl+g", "toggle_subagent_panel", "Toggle Subagents", show=False),
Binding(
"shift+tab",
"toggle_auto_approve",
@@ -2351,6 +2367,9 @@ class DeepAgentsApp(App):
)
yield Container(id="messages")
with Container(id="bottom-app-container"):
# Live fan-out panel for subagents spawned from js_eval. Hidden
# until the first spawn event; sits just above the input.
yield SubagentPanel(id="subagent-panel")
yield ChatInput(
cwd=self._cwd,
image_tracker=self._image_tracker,
@@ -2670,6 +2689,7 @@ class DeepAgentsApp(App):
sync_message_content=self._sync_message_content,
request_ask_user=self._request_ask_user,
on_tool_complete=self._schedule_git_branch_refresh,
on_subagent_event=self._on_subagent_event,
)
# Wire token display callbacks
self._ui_adapter._on_tokens_update = self._on_tokens_update
@@ -7283,6 +7303,10 @@ class DeepAgentsApp(App):
self._queued_widgets.clear()
self._sync_status_queued()
await self._clear_messages()
# A fresh conversation drops any prior subagent fan-out too.
subagent_panel = self._get_subagent_panel()
if subagent_panel is not None:
subagent_panel.reset()
self._context_tokens = 0
self._tokens_approximate = False
self._update_tokens(0)
@@ -8070,6 +8094,16 @@ class DeepAgentsApp(App):
turn_stats = SessionStats()
self._inflight_turn_stats = turn_stats
self._inflight_turn_start = time.monotonic()
# Arm the subagent fan-out panel for this turn, seeding the session
# model that labels each row. The panel persists across turns and only
# clears when this turn's first subagent actually starts, so a turn that
# spawns none leaves the previous workflow's results on screen.
panel = self._get_subagent_panel()
if panel is not None:
spec = self._effective_model_spec()
panel.prepare_turn(model_label=_display_model_label(spec))
try:
await execute_task_textual(
user_input=message,
@@ -8133,6 +8167,13 @@ class DeepAgentsApp(App):
if self._inflight_turn_stats is not None:
self._session_stats.merge(turn_stats)
self._inflight_turn_stats = None
# Finalize any subagent rows left "running" — an interrupt cancels
# the worker before the bridge emits terminal events (a cancel is a
# BaseException, which the bridge's `except Exception` skips), so the
# panel would otherwise spin forever. No-op when nothing's running.
subagent_panel = self._get_subagent_panel()
if subagent_panel is not None:
subagent_panel.finalize_running()
await self._cleanup_agent_task()
async def _process_next_from_queue(self) -> None:
@@ -9375,6 +9416,33 @@ class DeepAgentsApp(App):
restore_iterm_cursor_guide()
super().exit(result=result, return_code=return_code, message=message)
def _get_subagent_panel(self) -> SubagentPanel | None:
"""Return the subagent fan-out panel, or None if not yet mounted.
Returns:
The mounted `SubagentPanel`, or None during early startup.
"""
try:
return self.query_one("#subagent-panel", SubagentPanel)
except Exception: # noqa: BLE001 — not mounted during early startup
return None
def _on_subagent_event(self, event: dict[str, Any]) -> None:
"""Forward a validated subagent custom-stream event to the panel.
Runs on the Textual event loop (same loop as the stream consumer), so
the panel widget can be updated directly.
"""
panel = self._get_subagent_panel()
if panel is not None:
panel.on_subagent_event(event)
def action_toggle_subagent_panel(self) -> None:
"""Expand or collapse the subagent fan-out panel."""
panel = self._get_subagent_panel()
if panel is not None:
panel.toggle()
async def action_toggle_auto_approve(self) -> None:
"""Toggle auto-approve mode for the current session.
libs/code/deepagents_code/textual_adapter.py
+58 -1
View File
@@ -239,6 +239,7 @@ class TextualUIAdapter:
| None
) = None,
on_tool_complete: Callable[[], None] | None = None,
on_subagent_event: Callable[[dict[str, Any]], None] | None = None,
) -> None:
"""Initialize the adapter."""
self._mount_message = mount_message
@@ -281,6 +282,14 @@ class TextualUIAdapter:
for the full turn to finish.
"""
self._on_subagent_event = on_subagent_event
"""Sync callback fired for each validated `subagent` custom-stream event.
Drives the live subagent fan-out panel. Events originate from the
QuickJS `task()` bridge during a `js_eval` call; payload strings are
LLM/JS-authored and treated as untrusted by the panel renderer.
"""
# State tracking
self._current_tool_messages: dict[str, ToolCallMessage] = {}
"""Map of tool call IDs to their message widgets."""
@@ -374,6 +383,23 @@ def _read_mentioned_file(file_path: Path, max_embed_bytes: int) -> str:
return f"\n### {file_path.name}\nPath: `{file_path}`\n```\n{content}\n```"
def _is_renderable_subagent_event(data: Any, *, is_main_agent: bool) -> bool: # noqa: ANN401 # custom-stream payload is dynamic
"""Whether a `custom` payload is a subagent event this UI can render.
Guards the live panel against unrelated/malformed custom events and against
nested (subagent-to-subagent) emissions.
Args:
data: The `custom` stream payload.
is_main_agent: Whether the event came from the main agent's namespace
(the empty namespace). Nested emissions are ignored.
Returns:
True only for a well-formed subagent event from the main agent.
"""
return is_main_agent and isinstance(data, dict) and data.get("type") == "subagent"
async def execute_task_textual(
user_input: str,
agent: Any, # noqa: ANN401 # Dynamic agent graph type
@@ -582,7 +608,7 @@ async def execute_task_textual(
async for chunk in agent.astream(
stream_input,
stream_mode=["messages", "updates"],
stream_mode=["messages", "updates", "custom"],
subgraphs=True,
config=config,
context=context,
@@ -602,6 +628,25 @@ async def execute_task_textual(
# report back to the main agent
is_main_agent = ns_key == ()
# Handle CUSTOM stream - live subagent fan-out events emitted by
# the QuickJS task() bridge during a js_eval call. Validate at
# this boundary before forwarding so unrelated/malformed or
# nested custom events never reach the panel; forwarding must
# never raise into the stream loop.
if current_stream_mode == "custom":
if (
adapter._on_subagent_event is not None
and _is_renderable_subagent_event(
data, is_main_agent=is_main_agent
)
):
try:
adapter._on_subagent_event(data)
except Exception:
# Panel rendering must never crash the stream loop.
logger.exception("subagent panel event handler failed")
continue
# Handle UPDATES stream - for interrupts and todos
if current_stream_mode == "updates":
if not isinstance(data, dict):
@@ -1522,6 +1567,11 @@ async def _handle_interrupt_cleanup(
captured_output_tokens: Output tokens captured before interrupt.
turn_stats: Stats for the current turn.
start_time: Monotonic timestamp when the turn began.
Raises:
ValueError: If proactive remote-run cancellation is attempted without a
`thread_id` in `config` (a contract violation rather than a
transient remote failure).
"""
from langchain_core.messages import HumanMessage
@@ -1549,7 +1599,14 @@ async def _handle_interrupt_cleanup(
if cancel_active_runs is not None:
try:
await cancel_active_runs(config)
except ValueError:
# A missing thread_id is a contract violation (a bug), not a
# transient remote failure — surface it rather than downgrading it
# to a warning alongside the swallowed network errors below.
raise
except Exception:
# Remote cancel is best-effort defense-in-depth; transient remote
# failures here are recovered by aupdate_state's 409 retry below.
logger.warning(
"Failed to cancel active remote runs for thread %s",
config.get("configurable", {}).get("thread_id"),
libs/code/deepagents_code/widgets/subagent_panel.py
+969
View File
@@ -0,0 +1,969 @@
"""Live panel showing subagents fanned out from within `js_eval` calls.
When the agent writes code that calls the top-level `task()` global, each
dispatch runs as a subagent *inside* a single `js_eval` tool call which is
invisible to the normal message stream. The QuickJS task bridge emits
lifecycle events on the custom stream. This widget consumes them and renders
a docked, live-updating fan-out panel.
Trust note: `description`/`subagent_type` and `error` strings originate
from LLM-authored JavaScript executed in the sandbox, so they are untrusted.
We route every rendered string through `sanitize_control_chars` which strips
control/escape/bidi characters and only ever render via `Content.styled` /
`markup=False` `Static` updates, so embedded Textual markup and terminal
escapes cannot influence rendering or panel state.
"""
from __future__ import annotations
import contextlib
import logging
import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Literal
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.content import Content
from textual.css.query import NoMatches, TooManyMatches
from textual.reactive import reactive
from textual.widgets import Static
from deepagents_code.config import get_glyphs
from deepagents_code.formatting import format_duration
from deepagents_code.theme import get_theme_colors
from deepagents_code.unicode_security import sanitize_control_chars
from deepagents_code.widgets.loading import Spinner
if TYPE_CHECKING:
from textual import events
from textual.app import ComposeResult
from textual.timer import Timer
logger = logging.getLogger(__name__)
SubagentStatus = Literal["running", "done", "error", "cancelled"]
_MODEL_COL = 16
_TIMING_COL = 6
_STATUS_COL = 5
_MIN_TASK_COL = 16
_SCROLLBAR_RESERVE = 2
_FALLBACK_WIDTH = 100
_MIN_BODY_HEIGHT = 3
_MAX_BODY_HEIGHT = 12
_AGENTS_CHROME_LINES = 1
_TICK_INTERVAL = 0.1
_LABEL_FALLBACK_MAX_CHARS = 60
def _right_block_width() -> int:
"""Total width of the right-aligned metadata block (model→time).
Returns:
The combined character width of the model and time columns.
"""
gap = 2
return _MODEL_COL + gap + _TIMING_COL
@dataclass
class _SubagentRecord:
"""One subagent's live state within a phase."""
id: str
"""Per-dispatch subagent id from the stream event."""
label: str
"""Sanitized, display-ready task label for the row."""
status: SubagentStatus = "running"
"""Lifecycle state; starts running and moves to a terminal value once."""
started_monotonic: float = field(default_factory=time.monotonic)
"""Monotonic timestamp captured when the record was created."""
duration_ms: int | None = None
"""Measured duration once finished; None while still running."""
error: str | None = None
"""Failure reason, set only when status is error."""
def elapsed_seconds(self) -> float:
"""Seconds since this subagent started (live for running rows).
Returns:
The measured duration once finished, else the live elapsed time.
"""
if self.duration_ms is not None:
return self.duration_ms / 1000
return max(0.0, time.monotonic() - self.started_monotonic)
@dataclass
class _Phase:
"""One `js_eval` fan-out batch, keyed by the eval's tool-call id."""
eval_id: str
"""Parent `js_eval` tool-call id, or empty string when none was provided."""
index: int
"""1-based display ordinal assigned when the phase is created."""
records: dict[str, _SubagentRecord] = field(default_factory=dict)
"""Subagent records keyed by id; kept in sync with `order` via `add`."""
order: list[str] = field(default_factory=list)
"""Record ids in arrival order, defining render sequence."""
def add(self, record: _SubagentRecord) -> None:
"""Insert or replace a subagent record, preserving arrival order."""
if record.id not in self.records:
self.order.append(record.id)
self.records[record.id] = record
def counts(self) -> tuple[int, int]:
"""Return (finished, total) subagent counts for this phase."""
total = len(self.records)
done = sum(1 for r in self.records.values() if r.status != "running")
return done, total
def any_running(self) -> bool:
"""Whether any subagent in this phase is still running.
Returns:
True if at least one subagent has not finished.
"""
return any(r.status == "running" for r in self.records.values())
def any_error(self) -> bool:
"""Whether any subagent in this phase ended in error.
Returns:
True if at least one subagent ended in error.
"""
return any(r.status == "error" for r in self.records.values())
def any_cancelled(self) -> bool:
"""Whether any subagent in this phase was cancelled.
Returns:
True if at least one subagent was cancelled.
"""
return any(r.status == "cancelled" for r in self.records.values())
def all_terminal(self) -> bool:
"""Whether the phase has records and none are still running.
Returns:
True if the phase has at least one record and all have finished.
"""
return bool(self.records) and not self.any_running()
def elapsed_seconds(self) -> float:
"""Wall-clock elapsed for the phase (frozen once all subagents end).
Measured from the first subagent's start to the last one's finish, so
the value is continuous: the live "now - first start" simply freezes
when the final subagent ends (rather than collapsing to the longest
single duration).
Returns:
Live elapsed while running, else first-start to last-finish.
"""
if not self.records:
return 0.0
earliest = min(r.started_monotonic for r in self.records.values())
if self.all_terminal():
latest_end = max(
r.started_monotonic + r.elapsed_seconds() for r in self.records.values()
)
return max(0.0, latest_end - earliest)
return max(0.0, time.monotonic() - earliest)
def _format_timing(seconds: float) -> str:
"""Stable-width elapsed string for the table.
`format_duration` drops the decimal on whole seconds (`4s` vs `4.2s`),
which makes a live-ticking value jump left/right by a character each tick.
Always keep one decimal under a minute so the width stays constant.
Returns:
e.g. `4.0s` or `4.2s` under a minute, else `format_duration`'s output.
"""
if seconds < 60: # noqa: PLR2004
return f"{seconds:.1f}s"
return format_duration(seconds)
def _sanitize(text: str, *, max_chars: int) -> str:
"""Neutralize control/escape/bidi chars and bound length for a one-line label.
Inputs are LLM/JS-authored and untrusted. This flattens to a single line (newlines
and ANSI escapes become spaces) so a crafted description cannot inject terminal
escapes or extra rows.
Returns:
A single-line, length-bounded string safe to render as plain text.
"""
return sanitize_control_chars(text, keep_newlines=False, max_length=max_chars)
class SubagentPanel(Vertical):
"""Docked two-pane panel visualizing `js_eval` subagent fan-out by phase.
Hidden until the first spawn event. Phases (one per `js_eval`) list on the
left and the selected phase's subagents render as a scrollable table on the
right. Focus the panel and use up/down to revisit finished phases. Expands
while any phase runs, collapses to the header when the turn goes idle, and
re-expands when a new phase starts.
"""
can_focus = True
can_focus_children = False
DEFAULT_CSS = """
SubagentPanel {
height: auto;
background: $surface;
border-top: solid $primary;
display: none;
padding: 1 2;
}
SubagentPanel.-collapsed {
padding: 0 2;
}
SubagentPanel.-visible {
display: block;
}
SubagentPanel:focus {
border-top: solid $accent;
}
SubagentPanel #subagent-header {
width: 1fr;
height: 1;
text-style: bold;
}
SubagentPanel #subagent-body {
width: 1fr;
height: auto;
margin-top: 1;
}
SubagentPanel #subagent-body.-collapsed {
display: none;
}
SubagentPanel #subagent-phases-scroll {
width: 24;
height: 100%;
border-right: solid $primary-darken-2;
padding-right: 2;
margin-right: 2;
}
SubagentPanel #subagent-phases-scroll.-hidden {
display: none;
}
SubagentPanel #subagent-agents-scroll {
width: 1fr;
height: 100%;
}
"""
expanded: reactive[bool] = reactive(default=True, init=False)
def __init__(self, **kwargs: Any) -> None:
"""Initialize an empty, hidden panel."""
super().__init__(**kwargs)
self._phases: dict[str, _Phase] = {}
self._phase_order: list[str] = []
self._active_eval_id: str | None = None
self._selected_eval_id: str | None = None
self._model_label: str | None = None
self._applied_height: int | None = None
self._last_render: dict[str, str] = {}
self._spinner = Spinner()
self._timer: Timer | None = None
# When True, the next subagent `start` clears the previous workflow's
# fan-out before adding the new row (armed by `prepare_turn`). Lets the
# panel persist across turns until a new workflow actually begins.
self._pending_reset = False
def compose(self) -> ComposeResult: # noqa: PLR6301 — Textual widget method
"""Yield the header line and the two-pane body (phases | agents)."""
yield Static("", id="subagent-header", markup=False)
with Horizontal(id="subagent-body"):
with VerticalScroll(id="subagent-phases-scroll"):
yield Static("", id="subagent-phases", markup=False)
with VerticalScroll(id="subagent-agents-scroll"):
yield Static("", id="subagent-agents", markup=False)
@property
def _active_phase(self) -> _Phase | None:
if self._active_eval_id is None:
return self._phases.get("")
return self._phases.get(self._active_eval_id)
def _displayed_phase(self) -> _Phase | None:
"""The phase whose table is shown — the user's pick, else the active one.
Returns:
The selected phase if the user navigated to one, else the active
(latest) phase, or None when no phase has started.
"""
if self._selected_eval_id is not None:
phase = self._phases.get(self._selected_eval_id)
if phase is not None:
return phase
return self._active_phase
def on_subagent_event(self, event: dict[str, Any]) -> None:
"""Apply one validated subagent lifecycle event.
The caller (textual adapter) has already checked `type == "subagent"`
and that this is the main-agent namespace. We defensively re-validate
every field here so malformed payloads can never corrupt panel state.
"""
phase = event.get("phase")
sub_id = event.get("id")
if not isinstance(sub_id, str) or not sub_id:
# Producer/consumer contract drift — leave a breadcrumb rather than
# dropping the event with no trace.
logger.debug("Dropping subagent event with missing/invalid id: %r", event)
return
eval_id = event.get("eval_id")
eval_key = eval_id if isinstance(eval_id, str) else ""
if phase == "start":
self._handle_start(sub_id, eval_key, event)
elif phase in {"complete", "error"}:
self._handle_finish(sub_id, eval_key, phase, event)
else:
logger.debug(
"Dropping subagent event with unrecognized phase %r (id=%s)",
phase,
sub_id,
)
return
self._refresh()
def _handle_start(self, sub_id: str, eval_key: str, event: dict[str, Any]) -> None:
"""Create/replace a running record and (re-)show the panel."""
if self._pending_reset:
# A new workflow is starting — drop the previous turn's fan-out now.
self._clear()
phase = self._ensure_phase(eval_key)
self._active_eval_id = eval_key
record = _SubagentRecord(
id=sub_id,
label=_sanitize(self._row_label(event), max_chars=200),
)
phase.add(record)
self._show()
self._apply_body_height()
self._ensure_timer()
def _ensure_phase(self, eval_key: str) -> _Phase:
"""Return the phase for `eval_key`, creating and ordering it if new.
Returns:
The existing or newly created `_Phase` for this eval batch.
"""
phase = self._phases.get(eval_key)
if phase is None:
phase = _Phase(eval_id=eval_key, index=len(self._phase_order) + 1)
self._phases[eval_key] = phase
self._phase_order.append(eval_key)
return phase
@staticmethod
def _row_label(event: dict[str, Any]) -> str:
"""Build the row's task label: `"<type>: <label>"`.
Returns:
The combined `"<type>: <label>"` string for the task column.
"""
sub_type = event.get("subagent_type", "subagent")
label = event.get("label")
if not isinstance(label, str) or not label:
description = event.get("description")
label = description if isinstance(description, str) else ""
label = " ".join(label.split())[:_LABEL_FALLBACK_MAX_CHARS]
return f"{sub_type}: {label}"
def _handle_finish(
self, sub_id: str, eval_key: str, outcome: str, event: dict[str, Any]
) -> None:
"""Mark a record done/error, recording duration and stopping the timer.
An error with no matching `start` is adopted as a fresh row (see
`_adopt_orphan_finish`) so a dropped-start failure still surfaces.
"""
record = self._find_record(sub_id)
if record is None:
if outcome != "error":
# A success with no matching `start` has no row or label to
# attach to, so ignore it rather than creating a phantom phase.
# Log it, though: a dropped `start` is the same producer/consumer
# contract drift the other drop paths leave breadcrumbs for.
logger.debug(
"Dropping subagent complete event with no matching start "
"(id=%s, eval_id=%s)",
sub_id,
eval_key,
)
return
# An error with no matching `start` (e.g. the start event was
# dropped on the wire) carries the failure string — synthesize a
# minimal record so it still surfaces instead of vanishing silently.
record = self._adopt_orphan_finish(sub_id, eval_key, event)
record.status = "done" if outcome == "complete" else "error"
duration = event.get("duration_ms")
if isinstance(duration, (int, float)):
record.duration_ms = int(duration)
if outcome == "error":
raw_err = event.get("error")
record.error = (
_sanitize(raw_err, max_chars=120) if isinstance(raw_err, str) else None
)
if not self._any_running():
self._stop_timer()
def _adopt_orphan_finish(
self, sub_id: str, eval_key: str, event: dict[str, Any]
) -> _SubagentRecord:
"""Create a record for a terminal event that has no matching `start`.
Places it in the event's phase (creating the phase if needed) and shows
the panel so a dropped-start failure is still visible to the user.
Returns:
The newly created record, already inserted into its phase.
"""
if self._pending_reset:
# A dropped-start error is still evidence that a new workflow
# started, so clear the previous turn before surfacing it.
self._clear()
phase = self._ensure_phase(eval_key)
self._active_eval_id = eval_key
record = _SubagentRecord(
id=sub_id,
label=_sanitize(self._row_label(event), max_chars=200),
)
phase.add(record)
self._show()
self._apply_body_height()
return record
def _find_record(self, sub_id: str) -> _SubagentRecord | None:
"""Locate a record by id across all phases.
Returns:
The matching record, or None if no phase holds that id.
"""
for phase in self._phases.values():
record = phase.records.get(sub_id)
if record is not None:
return record
return None
def _any_running(self) -> bool:
"""Whether any phase still has a running subagent.
Returns:
True if any subagent in any phase is still running.
"""
return any(phase.any_running() for phase in self._phases.values())
def on_click(self, event: events.Click) -> None:
"""Click the header to toggle; click a phase row to select it."""
if self._header_clicked(event):
self.toggle()
event.stop()
return
if len(self._phase_order) <= 1:
return
row = self._clicked_phase_row(event)
if row is not None and 0 <= row < len(self._phase_order):
self._selected_eval_id = self._phase_order[row]
self._refresh()
event.stop()
def _header_clicked(self, event: events.Click) -> bool:
"""Whether the click landed on the header line.
Returns:
True if the click offset maps onto the header widget.
"""
try:
header = self.query_one("#subagent-header", Static)
except (NoMatches, TooManyMatches): # not mounted yet
return False
return event.get_content_offset(header) is not None
def _clicked_phase_row(self, event: events.Click) -> int | None:
"""Map a click in the phases pane to a phase index (0-based), or None.
Row 0 is the "Phases" title; rows 1..N map to phases in order.
Returns:
The 0-based phase index for the clicked row, or None if the click
was outside the phases pane.
"""
try:
phases = self.query_one("#subagent-phases", Static)
except (NoMatches, TooManyMatches): # not mounted yet
return None
offset = event.get_content_offset(phases)
if offset is None:
return None
return offset.y - 1
def on_key(self, event: events.Key) -> None:
"""Navigate phases with up/down (or j/k) while the panel is focused."""
if len(self._phase_order) <= 1:
return
if event.key in {"down", "j"}:
self._move_selection(1)
event.stop()
elif event.key in {"up", "k"}:
self._move_selection(-1)
event.stop()
def _move_selection(self, delta: int) -> None:
"""Move the selected phase by `delta`, clamped to the phase list."""
if not self._phase_order:
return
current = self._displayed_phase()
current_key = current.eval_id if current else self._phase_order[0]
try:
index = self._phase_order.index(current_key)
except ValueError:
index = 0
new_index = max(0, min(len(self._phase_order) - 1, index + delta))
self._selected_eval_id = self._phase_order[new_index]
self._refresh()
def prepare_turn(self, *, model_label: str | None = None) -> None:
"""Arm a deferred clear for a new turn without touching the panel yet.
The visible fan-out persists across turns; it is cleared lazily when the
next workflow actually starts a subagent (see `_handle_start`), so a turn
that spawns none leaves the previous results on screen. Refreshes the
session model used to label rows.
"""
self._model_label = (
_sanitize(model_label, max_chars=_MODEL_COL) if model_label else None
)
# If the previous turn left rows stuck "running" — e.g. it was cancelled
# before the bridge emitted terminal events (CancelledError is a
# BaseException, so it bypasses the bridge's `except Exception`) — clear
# now instead of persisting a stale, still-spinning fan-out. Otherwise
# defer the clear so a completed workflow survives no-subagent turns.
if self._any_running():
self._clear()
else:
self._pending_reset = True
def reset(self, *, model_label: str | None = None, **_kwargs: Any) -> None:
"""Clear all phases and hide the panel immediately (e.g. on `/clear`)."""
self._clear()
self._model_label = (
_sanitize(model_label, max_chars=_MODEL_COL) if model_label else None
)
def _clear(self) -> None:
"""Drop all phase state, stop the timer, and hide the panel.
Leaves `expanded` and `_model_label` untouched so the user's open/closed
choice and the session model survive a clear.
"""
self._phases.clear()
self._phase_order.clear()
self._active_eval_id = None
self._selected_eval_id = None
self._applied_height = None
self._last_render.clear()
self._pending_reset = False
self._stop_timer()
self.remove_class("-visible")
def finalize_running(self) -> None:
"""Mark any still-running subagents as cancelled and stop ticking.
Called when a turn is interrupted: the QuickJS bridge does not emit
terminal events for `asyncio.CancelledError` (a BaseException, so it
bypasses the bridge's `except Exception`), which would otherwise leave
rows spinning forever. Freezes each affected row's elapsed time.
"""
changed = False
for phase in self._phases.values():
for record in phase.records.values():
if record.status == "running":
record.status = "cancelled"
if record.duration_ms is None:
record.duration_ms = int(record.elapsed_seconds() * 1000)
changed = True
if not changed:
return
self._stop_timer()
self._refresh()
def _show(self) -> None:
"""Make the panel visible (idempotent)."""
self.add_class("-visible")
def toggle(self) -> None:
"""Toggle the body open/closed. This is the only thing that changes it."""
self.expanded = not self.expanded
def watch_expanded(self, expanded: bool) -> None:
"""Show/hide the body when the expanded state changes."""
try:
body = self.query_one("#subagent-body")
except (NoMatches, TooManyMatches): # body not mounted yet
return
body.set_class(not expanded, "-collapsed")
# Drop vertical padding when collapsed so the header bar is thin.
self.set_class(not expanded, "-collapsed")
if expanded:
self._apply_body_height()
self._refresh()
# On re-expand the body was just shown (it had zero width while
# collapsed), so the agents pane width isn't known yet. Re-render once
# layout settles so the right-aligned columns use the real width.
if expanded:
self.call_after_refresh(self._refresh)
def on_resize(self, _event: events.Resize) -> None:
"""Re-render so width-dependent column alignment tracks the new size."""
self._refresh()
def _ensure_timer(self) -> None:
"""Start the refresh timer if it isn't already running."""
if self._timer is None:
self._timer = self.set_interval(_TICK_INTERVAL, self._refresh)
def _stop_timer(self) -> None:
"""Stop and drop the refresh timer if running."""
if self._timer is not None:
self._timer.stop()
self._timer = None
def _counts(self) -> tuple[int, int]:
"""(finished, total) for the displayed phase.
Returns:
A `(finished, total)` count for the displayed phase, or `(0, 0)`.
"""
phase = self._displayed_phase()
return phase.counts() if phase else (0, 0)
def _turn_counts(self) -> tuple[int, int, int, int]:
"""Sum subagent counts across all phases.
Returns:
A `(finished, total, failed, cancelled)` tuple over the whole turn.
"""
total = done = failed = cancelled = 0
for phase in self._phases.values():
for record in phase.records.values():
total += 1
if record.status != "running":
done += 1
if record.status == "error":
failed += 1
elif record.status == "cancelled":
cancelled += 1
return done, total, failed, cancelled
def _body_height(self) -> int:
"""Constant body height for the turn — sized to the largest phase.
Returns:
A cell height clamped to the configured bounds, stable across phase
switches.
"""
if not self._phases:
return _MIN_BODY_HEIGHT
max_rows = max(len(p.records) for p in self._phases.values())
agents_lines = _AGENTS_CHROME_LINES + max_rows
phases_lines = 1 + len(self._phases) # "Phases" title + one per phase
return min(_MAX_BODY_HEIGHT, max(_MIN_BODY_HEIGHT, agents_lines, phases_lines))
def _apply_body_height(self) -> None:
"""Lock the body to a constant height; only re-assign when it changes.
Re-assigning `styles.height` every timer tick forces a relayout and
causes visible flicker, so we cache the applied value and only set it
when the phase composition actually changes the needed height.
"""
if not self.expanded:
return
height = self._body_height()
if height == self._applied_height:
return
with contextlib.suppress(NoMatches, TooManyMatches): # not mounted yet
self.query_one("#subagent-body").styles.height = height
self._applied_height = height
def _update_cached(self, widget_id: str, content: Content) -> None:
"""Update a Static only when its rendered text changed (anti-flicker)."""
if self._last_render.get(widget_id) == content.plain:
return
try:
self.query_one(f"#{widget_id}", Static).update(content)
except (NoMatches, TooManyMatches): # not mounted yet
return
self._last_render[widget_id] = content.plain
def _refresh(self) -> None:
"""Re-render all three regions (header, phases pane, agents pane)."""
self._refresh_header()
self._refresh_phases()
self._refresh_agents()
def _refresh_header(self) -> None:
"""Render the header: status icon, label, whole-turn totals, toggle hint."""
colors = get_theme_colors(self)
glyphs = get_glyphs()
caret = (
glyphs.disclosure_expanded if self.expanded else glyphs.disclosure_collapsed
)
done, total, failed, cancelled = self._turn_counts()
if self._any_running() or not total:
icon, tint = self._spinner.next_frame(), colors.warning
elif failed:
icon, tint = glyphs.error, colors.error
elif cancelled:
icon, tint = glyphs.circle_empty, colors.muted
else:
icon, tint = glyphs.checkmark, colors.success
lead_text = f"{caret} {icon} dynamic subagents"
parts: list[Content] = [Content.styled(lead_text, tint)]
left_len = len(lead_text)
if self.expanded and total:
meta = self._header_meta_parts(done, total, failed, cancelled, colors)
parts.extend(meta)
left_len += sum(len(p.plain) for p in meta)
hint = (
"click or Ctrl+G to collapse"
if self.expanded
else "click or Ctrl+G to expand"
)
spacer = max(2, self._header_width() - left_len - len(hint))
parts.append(Content.styled(" " * spacer + hint, colors.muted))
self._update_cached("subagent-header", Content.assemble(*parts))
def _header_meta_parts(
self,
done: int,
total: int,
failed: int,
cancelled: int,
colors: Any, # noqa: ANN401 — ThemeColors
) -> list[Content]:
"""Whole-turn totals (phase count, failures, cancellations) when expanded.
Returns:
The styled `Content` pieces appended after the header label.
"""
meta_text = f" {done}/{total} done"
count = len(self._phase_order)
if count:
plural = "phase" if count == 1 else "phases"
meta_text += f" · {count} {plural}"
parts: list[Content] = [Content.styled(meta_text, colors.muted)]
if failed:
parts.append(Content.styled(f" · {failed} failed", colors.error))
if cancelled:
parts.append(Content.styled(f" · {cancelled} cancelled", colors.muted))
return parts
def _header_width(self) -> int:
"""Current cell width of the header line (fallback until laid out).
Returns:
The header width, or a fallback before first layout.
"""
try:
width = self.query_one("#subagent-header", Static).size.width
except (NoMatches, TooManyMatches): # not mounted yet
width = 0
return width if width and width > 0 else _FALLBACK_WIDTH
def _refresh_phases(self) -> None:
"""Render the left pane: one selectable row per phase (eval batch)."""
try:
scroll = self.query_one("#subagent-phases-scroll")
except (NoMatches, TooManyMatches): # not mounted yet
return
# Hide only when there are no phases at all; otherwise always show the
# list (even a single phase) for a consistent two-pane view.
if not self._phase_order:
scroll.add_class("-hidden")
self._update_cached("subagent-phases", Content(""))
return
scroll.remove_class("-hidden")
colors = get_theme_colors(self)
displayed = self._displayed_phase()
displayed_key = displayed.eval_id if displayed else None
rows: list[Content] = [Content.styled("Phases", colors.muted)]
rows.extend(
self._phase_row(
self._phases[key], selected=key == displayed_key, colors=colors
)
for key in self._phase_order
)
self._update_cached("subagent-phases", Content("\n").join(rows))
def _phase_row(
self,
phase: _Phase,
*,
selected: bool,
colors: Any, # noqa: ANN401 — ThemeColors
) -> Content:
"""Render one phase row: caret, status glyph, index, counts, elapsed.
Returns:
The styled `Content` for the phase's row in the left pane.
"""
glyphs = get_glyphs()
done, total = phase.counts()
if phase.all_terminal():
if phase.any_error():
mark = glyphs.error
elif phase.any_cancelled():
mark = glyphs.circle_empty
else:
mark = glyphs.checkmark
elif phase.eval_id == self._active_eval_id:
mark = glyphs.disclosure_collapsed
else:
mark = glyphs.bullet
caret = glyphs.cursor if selected else " "
tint = colors.primary if selected else colors.muted
elapsed = _format_timing(phase.elapsed_seconds())
return Content.styled(
f"{caret} {mark} {phase.index} {done}/{total} · {elapsed}", tint
)
def _agents_width(self) -> int:
"""Usable width of the agents pane (fallback until laid out).
Returns:
The agents pane width minus a scrollbar reserve, floored at the
minimum task width.
"""
try:
width = self.query_one("#subagent-agents", Static).size.width
except (NoMatches, TooManyMatches): # not mounted yet
width = 0
if not width or width <= 0:
width = _FALLBACK_WIDTH
# Keep the flush-right column off the scroll bar.
return max(_MIN_TASK_COL, width - _SCROLLBAR_RESERVE)
def _task_col(self) -> int:
"""Width of the flexible task column so the right block sits flush-right.
Returns:
The task column width, clamped to a sensible minimum.
"""
width = self._agents_width()
return max(_MIN_TASK_COL, width - _STATUS_COL - _right_block_width())
def _refresh_agents(self) -> None:
"""Render the right pane: heading + one row per subagent in the phase."""
phase = self._displayed_phase()
glyphs = get_glyphs()
colors = get_theme_colors(self)
rows: list[Content] = []
if phase is not None and phase.order:
task_col = self._task_col()
rows.append(self._heading_row(task_col, colors))
rows.extend(
self._render_row(phase.records[sub_id], task_col, glyphs, colors)
for sub_id in phase.order
)
self._update_cached(
"subagent-agents", Content("\n").join(rows) if rows else Content("")
)
@staticmethod
def _right_block(model: str, timing: str) -> str:
"""Format the fixed-width, flush-right metadata columns (model, time).
Returns:
The concatenated, column-aligned metadata string.
"""
return (
f"{model[:_MODEL_COL].ljust(_MODEL_COL)} "
f"{timing[:_TIMING_COL].rjust(_TIMING_COL)}"
)
def _render_row(
self,
record: _SubagentRecord,
task_col: int,
glyphs: Any, # noqa: ANN401 — Glyphs
colors: Any, # noqa: ANN401 — ThemeColors
) -> Content:
"""Render one row: status | task (left) | model · time (right).
On failure the reason is appended to the (wide) task column rather than
the 6-char time column, so it stays legible instead of being truncated.
Returns:
The styled, width-filling row `Content`.
"""
if record.status == "running":
icon = self._spinner.current_frame()
tint = colors.warning
elif record.status == "done":
icon = glyphs.checkmark
tint = colors.success
elif record.status == "cancelled":
icon = glyphs.circle_empty
tint = colors.muted
else:
icon = glyphs.error
tint = colors.error
label = record.label
if record.status == "error" and record.error:
label = f"{record.label} - {record.error}"
timing = _format_timing(record.elapsed_seconds())
task = _sanitize(label, max_chars=task_col - 1).ljust(task_col)
model = self._model_label or ""
right = self._right_block(model, timing)
return Content.assemble(
Content.styled(f" {icon} ", tint),
Content.styled(task, colors.foreground),
Content.styled(right, colors.muted),
)
def _heading_row(
self,
task_col: int,
colors: Any, # noqa: ANN401 — ThemeColors
) -> Content:
"""Column heading row aligned to the data rows.
Returns:
A single dim heading line spanning the same columns as the rows.
"""
prefix = " " * _STATUS_COL
task = "TASK".ljust(task_col)
right = self._right_block("MODEL", "TIME")
return Content.styled(f"{prefix}{task}{right}", colors.muted)
libs/code/tests/unit_tests/test_app.py
+24
View File
@@ -43,6 +43,7 @@ from deepagents_code.app import (
QueuedMessage,
TextualSessionState,
_build_whats_new_message,
_display_model_label,
_extra_is_ready,
)
from deepagents_code.event_bus import ExternalEvent
@@ -80,6 +81,29 @@ def _closing_run_worker_mock(
return MagicMock()
class TestDisplayModelLabel:
"""Tests for stripping the provider prefix off a model spec for display."""
@pytest.mark.parametrize(
("spec", "expected"),
[
("anthropic:opus", "opus"),
("openai:gpt-5.1", "gpt-5.1"),
# No prefix: shown verbatim.
("opus", "opus"),
# Only the first colon splits, so a colon in the model name survives.
("anthropic:claude:opus", "claude:opus"),
# Falsy specs pass through unchanged rather than raising.
("", ""),
(None, None),
],
)
def test_strips_provider_prefix(
self, spec: str | None, expected: str | None
) -> None:
assert _display_model_label(spec) == expected
class TestWhatsNewMessage:
"""Tests for the post-upgrade banner content."""
libs/code/tests/unit_tests/test_subagent_panel.py
+549
View File
@@ -0,0 +1,549 @@
"""Behavioral tests for the SubagentPanel widget.
Each test mounts the panel in a minimal App, feeds it realistic subagent
lifecycle events, and asserts on rendered content / observable state — not on
types. Uses the Textual `run_test()` pilot harness.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, cast
import pytest
from textual.app import App, ComposeResult
from textual.geometry import Offset
from textual.widgets import Static
from deepagents_code.widgets.subagent_panel import (
SubagentPanel,
_Phase,
_SubagentRecord,
)
if TYPE_CHECKING:
from typing import Any
class PanelApp(App):
"""Minimal app that mounts a SubagentPanel for testing."""
def compose(self) -> ComposeResult:
yield SubagentPanel(id="panel")
class _FakeClick:
"""Stand-in for a Textual Click that reports an offset for one target id."""
def __init__(self, *, row_y: int, target_id: str) -> None:
self._y = row_y
self._target = target_id
self.stopped = False
def get_content_offset(self, widget: object) -> Offset | None:
if getattr(widget, "id", None) == self._target:
return Offset(0, self._y)
return None
def stop(self) -> None:
self.stopped = True
def _start(
sub_id: str, eval_id: str, desc: str = "task", label: str | None = "work"
) -> dict:
event = {
"type": "subagent",
"phase": "start",
"id": sub_id,
"eval_id": eval_id,
"subagent_type": "research",
"description": desc,
}
if label is not None:
event["label"] = label
return event
def _complete(sub_id: str, eval_id: str, duration_ms: int = 100) -> dict:
return {
"type": "subagent",
"phase": "complete",
"id": sub_id,
"eval_id": eval_id,
"duration_ms": duration_ms,
}
def _error(sub_id: str, eval_id: str, message: str = "boom") -> dict:
return {
"type": "subagent",
"phase": "error",
"id": sub_id,
"eval_id": eval_id,
"duration_ms": 50,
"error": message,
}
def _render(widget: Static) -> str:
content = widget.render()
plain = getattr(content, "plain", None)
return plain if isinstance(plain, str) else str(content)
def _displayed_id(panel: SubagentPanel) -> str:
phase = panel._displayed_phase()
assert phase is not None
return phase.eval_id
class TestLifecycle:
async def test_hidden_until_first_spawn(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
assert not panel.has_class("-visible")
async def test_visible_and_expanded_after_spawn(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
await pilot.pause()
assert panel.has_class("-visible")
assert panel.expanded is True
assert panel._counts() == (0, 1)
async def test_any_running_tracks_terminal_state(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_start("b", "E1"))
await pilot.pause()
assert panel._any_running() is True
panel.on_subagent_event(_complete("a", "E1"))
panel.on_subagent_event(_complete("b", "E1"))
await pilot.pause()
assert panel._any_running() is False
assert panel._counts() == (2, 2)
async def test_missing_label_falls_back_to_short_description(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
description = "Review\n" + "a" * 100
panel.on_subagent_event(_start("a", "E1", desc=description, label=None))
await pilot.pause()
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert ("Review " + "a" * 100)[:60] in rows
async def test_error_shows_full_reason_in_task_column(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1", label="db.ts"))
panel.on_subagent_event(_error("a", "E1", message="rate limit exceeded"))
await pilot.pause()
assert panel._any_running() is False
record = panel._find_record("a")
assert record is not None
assert record.status == "error"
assert record.error == "rate limit exceeded"
# The full reason appears in the wide task column; it would be cut to
# ~6 chars if it were rendered in the (narrow) TIME column.
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "rate limit exceeded" in rows
async def test_orphan_error_surfaces_without_start(self) -> None:
# An error whose `start` was dropped on the wire must still surface as a
# failed row rather than vanishing silently.
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_error("orphan", "E1", message="dropped boom"))
await pilot.pause()
assert panel.has_class("-visible")
record = panel._find_record("orphan")
assert record is not None
assert record.status == "error"
assert record.error == "dropped boom"
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "dropped boom" in rows
async def test_orphan_error_after_prepare_turn_replaces_prior_turn(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1", label="old work"))
panel.on_subagent_event(_complete("a", "E1"))
panel.prepare_turn()
panel.on_subagent_event(_error("orphan", "E2", message="dropped boom"))
panel.on_subagent_event(_start("b", "E3", label="later work"))
await pilot.pause()
assert panel._phase_order == ["E2", "E3"]
assert panel._find_record("a") is None
assert panel._find_record("orphan") is not None
assert panel._find_record("b") is not None
async def test_orphan_error_becomes_active_phase(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1", label="old work"))
panel.on_subagent_event(_complete("a", "E1"))
panel.on_subagent_event(_error("orphan", "E2", message="dropped boom"))
await pilot.pause()
assert _displayed_id(panel) == "E2"
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "dropped boom" in rows
async def test_orphan_error_without_duration_still_renders(self) -> None:
# The realistic dropped-wire case: a partial error event missing
# `duration_ms` must still surface, leaving the duration unset rather
# than crashing on the missing/non-numeric field.
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(
{
"type": "subagent",
"phase": "error",
"id": "orphan",
"eval_id": "E1",
"error": "dropped boom",
}
)
await pilot.pause()
record = panel._find_record("orphan")
assert record is not None
assert record.status == "error"
assert record.duration_ms is None
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "dropped boom" in rows
class TestPhases:
async def test_phases_accumulate_and_track_active(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_complete("a", "E1"))
panel.on_subagent_event(_start("b", "E2"))
await pilot.pause()
assert panel._phase_order == ["E1", "E2"]
assert panel._active_eval_id == "E2"
# Earlier phase retained; active table shows only the new phase.
assert set(panel._phases["E1"].records) == {"a"}
assert _displayed_id(panel) == "E2"
async def test_eval_without_subagents_creates_no_phase(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
# A complete with no prior start is a no-op (no phantom phase).
panel.on_subagent_event(_complete("ghost", "E9"))
await pilot.pause()
assert panel._phase_order == []
assert not panel.has_class("-visible")
async def test_missing_eval_id_groups_into_single_phase(self) -> None:
# When the runtime exposes no tool_call_id the producer omits `eval_id`;
# such events share the empty-string phase key. Document that collapse so
# a future change that needs to distinguish them is forced to notice.
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
for sub_id in ("a", "b"):
panel.on_subagent_event(
{
"type": "subagent",
"phase": "start",
"id": sub_id,
"subagent_type": "research",
"description": "task",
"label": "work",
}
)
await pilot.pause()
assert panel._phase_order == [""]
assert set(panel._phases[""].records) == {"a", "b"}
class TestSelection:
async def test_selection_follows_active_then_locks_on_navigation(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1", label="phase one work"))
panel.on_subagent_event(_complete("a", "E1"))
panel.on_subagent_event(_start("b", "E2", label="phase two work"))
await pilot.pause()
assert _displayed_id(panel) == "E2"
panel._move_selection(-1)
await pilot.pause()
assert _displayed_id(panel) == "E1"
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "phase one work" in rows
async def test_selection_clamped_at_bounds(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_start("b", "E2"))
await pilot.pause()
panel._move_selection(-5) # past the top
assert _displayed_id(panel) == "E1"
panel._move_selection(5) # past the bottom
assert _displayed_id(panel) == "E2"
async def test_click_selects_phase_row(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_start("b", "E2"))
await pilot.pause()
# Row 1 is the first phase (row 0 is the "Phases" title).
panel.on_click(
cast("Any", _FakeClick(row_y=1, target_id="subagent-phases"))
)
await pilot.pause()
assert _displayed_id(panel) == "E1"
class TestAgentsTable:
async def test_row_shows_label_not_description(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(
_start("a", "E1", desc="a long boilerplate prompt", label="R16: #3")
)
await pilot.pause()
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "R16: #3" in rows # the label is what's rendered
assert "boilerplate" not in rows # the description never reaches the row
async def test_rows_show_session_model_label(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.reset(model_label="opus")
panel.on_subagent_event(_start("a", "E1", label="x"))
await pilot.pause()
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "opus" in rows
async def test_rows_show_headings(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1", label="x"))
await pilot.pause()
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "TASK" in rows
assert "MODEL" in rows
assert "TIME" in rows
class TestHeaderToggle:
async def test_toggle_flips_expanded(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
await pilot.pause()
assert panel.expanded is True
panel.toggle()
await pilot.pause()
assert panel.expanded is False
async def test_user_collapse_persists_across_turn_reset(self) -> None:
async with PanelApp().run_test(size=(160, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
await pilot.pause()
panel.toggle() # user closes it
panel.reset() # new user turn
panel.on_subagent_event(_start("b", "E2"))
await pilot.pause()
assert panel.expanded is False # preference persists
async def test_header_shows_turn_totals_and_failed(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_start("b", "E1"))
panel.on_subagent_event(_complete("a", "E1"))
panel.on_subagent_event(_error("b", "E1"))
await pilot.pause()
header = _render(pilot.app.query_one("#subagent-header", Static))
assert "2/2 done" in header
assert "1 phase" in header # singular for a single phase
assert "1 failed" in header
async def test_header_phase_count_pluralizes(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
await pilot.pause()
header = _render(pilot.app.query_one("#subagent-header", Static))
assert "1 phase" in header
assert "1 phases" not in header # singular, not "1 phases"
# A second eval batch makes it plural.
panel.on_subagent_event(_start("b", "E2"))
await pilot.pause()
header = _render(pilot.app.query_one("#subagent-header", Static))
assert "2 phases" in header
class TestReset:
async def test_reset_hides_and_clears(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
await pilot.pause()
panel.reset()
await pilot.pause()
assert not panel.has_class("-visible")
assert panel._phase_order == []
assert panel._counts() == (0, 0)
async def test_panel_persists_until_next_workflow(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_complete("a", "E1"))
await pilot.pause()
assert panel.has_class("-visible")
# A new turn begins but spawns no subagents — results persist.
panel.prepare_turn()
await pilot.pause()
assert panel.has_class("-visible")
assert panel._phase_order == ["E1"]
# The next workflow's first subagent clears the prior fan-out.
panel.on_subagent_event(_start("b", "E2"))
await pilot.pause()
assert panel._phase_order == ["E2"]
assert panel._find_record("a") is None
async def test_finalize_running_marks_cancelled(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_start("b", "E1"))
await pilot.pause()
assert panel._any_running() is True
# The turn is interrupted — finalize the in-flight rows.
panel.finalize_running()
await pilot.pause()
assert panel._any_running() is False
rec_a = panel._find_record("a")
rec_b = panel._find_record("b")
assert rec_a is not None
assert rec_b is not None
assert rec_a.status == "cancelled"
assert rec_b.status == "cancelled"
header = _render(pilot.app.query_one("#subagent-header", Static))
assert "2 cancelled" in header
async def test_finalize_running_preserves_finished_rows(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1"))
panel.on_subagent_event(_complete("a", "E1"))
panel.on_subagent_event(_start("b", "E1")) # still running
await pilot.pause()
panel.finalize_running()
await pilot.pause()
rec_a = panel._find_record("a")
rec_b = panel._find_record("b")
assert rec_a is not None
assert rec_b is not None
assert rec_a.status == "done" # already finished — untouched
assert rec_b.status == "cancelled" # in-flight — cancelled
async def test_prepare_turn_clears_stuck_running_rows(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
# A subagent starts but never finishes (e.g. the turn was cancelled
# before a terminal event arrived — CancelledError bypasses the
# bridge's terminal-event emission).
panel.on_subagent_event(_start("a", "E1"))
await pilot.pause()
assert panel._any_running() is True
# The next turn must not persist a stale, still-running fan-out.
panel.prepare_turn()
await pilot.pause()
assert not panel.has_class("-visible")
assert panel._phase_order == []
class TestStability:
async def test_body_height_stable_across_phase_switch(self) -> None:
async with PanelApp().run_test(size=(160, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
# Phase 1 has 3 subagents; phase 2 has 1.
for sid in ("a", "b", "c"):
panel.on_subagent_event(_start(sid, "E1"))
panel.on_subagent_event(_complete(sid, "E1"))
panel.on_subagent_event(_start("d", "E2"))
await pilot.pause()
height_active = panel._body_height()
panel._move_selection(-1) # show the smaller phase 1
await pilot.pause()
assert panel._body_height() == height_active # sized to largest phase
async def test_idle_refresh_skips_redundant_agent_update(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
async with PanelApp().run_test(size=(160, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event(_start("a", "E1", label="x"))
panel.on_subagent_event(_complete("a", "E1"))
await pilot.pause()
agents = pilot.app.query_one("#subagent-agents", Static)
calls = {"n": 0}
def _counting(*_args: object, **_kwargs: object) -> None:
calls["n"] += 1
monkeypatch.setattr(agents, "update", _counting)
panel._refresh() # nothing changed since last render
assert calls["n"] == 0
class TestSafety:
async def test_ignored_events_are_noops(self) -> None:
async with PanelApp().run_test() as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
panel.on_subagent_event({"phase": "start"}) # no id
panel.on_subagent_event({"phase": "weird", "id": "a"}) # bad phase
await pilot.pause()
assert panel._phase_order == []
assert not panel.has_class("-visible")
async def test_strips_escapes_and_bounds_length(self) -> None:
async with PanelApp().run_test(size=(200, 24)) as pilot:
panel = pilot.app.query_one("#panel", SubagentPanel)
nasty = "evil\x1b[31m\nsecond line"
panel.on_subagent_event(_start("a", "E1", label=nasty))
await pilot.pause()
rows = _render(pilot.app.query_one("#subagent-agents", Static))
assert "\x1b" not in rows # escape stripped
# The data row is a single line (newline flattened to a space).
data_row = rows.split("\n")[1]
assert "\n" not in data_row
assert "second line" in data_row
class TestPhaseTiming:
def test_phase_elapsed_is_wall_clock_not_longest_subagent(self) -> None:
# Two subagents with staggered starts, each running 3s:
# A: starts 100.0, ends 103.0
# B: starts 102.0, ends 105.0
# Wall-clock span is 5.0s (100.0 -> 105.0), not the 3.0s longest run.
phase = _Phase(eval_id="E1", index=1)
phase.add(
_SubagentRecord(
id="a",
label="a",
status="done",
started_monotonic=100.0,
duration_ms=3000,
)
)
phase.add(
_SubagentRecord(
id="b",
label="b",
status="done",
started_monotonic=102.0,
duration_ms=3000,
)
)
assert phase.elapsed_seconds() == pytest.approx(5.0)
libs/code/tests/unit_tests/test_subagent_stream.py
+41
View File
@@ -0,0 +1,41 @@
"""Unit tests for the subagent custom-stream boundary filter.
`_is_renderable_subagent_event` is the trust boundary between the agent's
`custom` stream and the live panel: only well-formed subagent events from the
main agent's namespace are forwarded.
"""
from __future__ import annotations
from deepagents_code.textual_adapter import _is_renderable_subagent_event
def _event(**overrides: object) -> dict:
event = {"type": "subagent", "id": "a"}
event.update(overrides)
return event
def test_accepts_well_formed_main_agent_event() -> None:
assert _is_renderable_subagent_event(_event(), is_main_agent=True) is True
def test_rejects_nested_namespace() -> None:
# Subagent-to-subagent emissions (non-empty namespace) are ignored.
assert _is_renderable_subagent_event(_event(), is_main_agent=False) is False
def test_rejects_non_subagent_custom_payload() -> None:
# Unrelated custom events (some other producer) must not reach the panel.
assert (
_is_renderable_subagent_event({"type": "progress"}, is_main_agent=True) is False
)
def test_rejects_payload_without_type() -> None:
assert _is_renderable_subagent_event({"id": "a"}, is_main_agent=True) is False
def test_rejects_non_dict_payload() -> None:
assert _is_renderable_subagent_event("nope", is_main_agent=True) is False
assert _is_renderable_subagent_event(None, is_main_agent=True) is False
libs/code/tests/unit_tests/test_textual_adapter.py
+35
View File
@@ -336,6 +336,41 @@ class TestInterruptCleanup:
agent.acancel_active_runs.assert_awaited_once()
agent.aupdate_state.assert_awaited_once()
async def test_remote_run_cancel_value_error_propagates(self) -> None:
"""A `ValueError` (missing `thread_id`) propagates instead of warning.
It is a contract bug rather than a transient remote failure, so it must
surface and the recovery-state write must be skipped.
"""
agent = SimpleNamespace(
acancel_active_runs=AsyncMock(side_effect=ValueError("missing thread_id")),
aupdate_state=AsyncMock(),
)
adapter = TextualUIAdapter(
mount_message=AsyncMock(),
update_status=_noop_status,
request_approval=_mock_approval,
set_spinner=AsyncMock(),
set_active_message=MagicMock(),
)
with pytest.raises(ValueError, match="missing thread_id"):
await _handle_interrupt_cleanup(
adapter=adapter,
agent=agent,
config={"configurable": {"thread_id": "t-1"}},
pending_text_by_namespace={},
captured_input_tokens=0,
captured_output_tokens=0,
turn_stats=SessionStats(),
start_time=0.0,
)
agent.acancel_active_runs.assert_awaited_once()
# The re-raise short-circuits before the recovery-state write, which
# is what distinguishes it from the swallowed-transient-failure path.
agent.aupdate_state.assert_not_awaited()
async def test_local_agent_without_cancel_method_still_writes_state(self) -> None:
"""Local agents lack `acancel_active_runs`; cleanup must skip it cleanly."""
agent = SimpleNamespace(aupdate_state=AsyncMock())
libs/partners/quickjs/langchain_quickjs/__init__.py
+12
View File
@@ -1,9 +1,21 @@
"""langchain-quickjs: persistent JS REPL middleware for agents."""
from langchain_quickjs._ptc import PTCOption
from langchain_quickjs._subagent import (
SUBAGENT_STREAM_EVENT_TYPE,
SubagentCompleteEvent,
SubagentErrorEvent,
SubagentStartEvent,
SubagentStreamEvent,
)
from langchain_quickjs.middleware import CodeInterpreterMiddleware
__all__ = [
"SUBAGENT_STREAM_EVENT_TYPE",
"CodeInterpreterMiddleware",
"PTCOption",
"SubagentCompleteEvent",
"SubagentErrorEvent",
"SubagentStartEvent",
"SubagentStreamEvent",
]
libs/partners/quickjs/langchain_quickjs/_prompt.py
+9 -2
View File
@@ -45,6 +45,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -62,6 +63,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -236,8 +241,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/langchain_quickjs/_ptc.py
+2 -2
View File
@@ -39,8 +39,8 @@ _RESERVED_SUBAGENT_TASK_NAME = "task"
_TASK_IN_PTC_MSG = (
"The subagent `task` tool cannot be exposed via `ptc`. It is always "
"available as the top-level `task()` global inside the REPL (with "
"`subagentType` and `responseSchema` support); exposing it through the "
"`tools.*` namespace would create a second, conflicting dispatch path "
"`subagentType`, `label`, and `responseSchema` support); exposing it through "
"the `tools.*` namespace would create a second, conflicting dispatch path "
'that drops `responseSchema`. Remove "task" from `ptc`.'
)
libs/partners/quickjs/langchain_quickjs/_repl.py
+38 -16
View File
@@ -531,6 +531,41 @@ class _ThreadREPL:
self._active_tool_names = target_names
self._tools_installed = True
@staticmethod
def _validate_task_payload(
payload: dict[str, Any],
) -> tuple[str, str, str | None, dict[str, Any] | None]:
"""Validate JS `task()` input and return its typed fields.
JS callers pass camelCase keys (`subagentType`, `responseSchema`) as
documented in the system prompt; the returned tuple is snake_case for
the Python dispatch path.
"""
description = payload.get("description")
if not isinstance(description, str) or not description:
msg = "task() requires non-empty string field `description`"
raise ValueError(msg)
subagent_type = payload.get("subagentType")
if not isinstance(subagent_type, str) or not subagent_type:
msg = "task() requires non-empty string field `subagentType`"
raise ValueError(msg)
raw_label = payload.get("label")
if raw_label is not None and not isinstance(raw_label, str):
msg = "task() field `label` must be a string when provided"
raise ValueError(msg)
label = raw_label.strip() if isinstance(raw_label, str) else None
if label == "":
label = None
response_schema = payload.get("responseSchema")
if response_schema is not None and not isinstance(response_schema, dict):
msg = "task() field `responseSchema` must be an object when provided"
raise ValueError(msg)
return description, subagent_type, label, response_schema
async def _ainvoke_task_on_outer_loop(
self,
payload: dict[str, Any],
@@ -543,22 +578,8 @@ class _ThreadREPL:
should execute on the parent LangGraph loop when one exists so callbacks,
context, and async loop affinity match normal tool execution.
"""
description = payload.get("description")
if not isinstance(description, str) or not description:
msg = "task() requires non-empty string field `description`"
raise ValueError(msg)
# JS callers use camelCase keys (`subagentType`, `responseSchema`) as
# documented in the system prompt.
subagent_type = payload.get("subagentType")
if not isinstance(subagent_type, str) or not subagent_type:
msg = "task() requires non-empty string field `subagentType`"
raise ValueError(msg)
response_schema = payload.get("responseSchema")
if response_schema is not None and not isinstance(response_schema, dict):
msg = "task() field `responseSchema` must be an object when provided"
raise ValueError(msg)
validated = self._validate_task_payload(payload)
description, subagent_type, label, response_schema = validated
async def _call() -> Any:
runtime = state.outer_runtime
@@ -575,6 +596,7 @@ class _ThreadREPL:
subagent_type=subagent_type,
response_schema=response_schema,
runtime=runtime,
label=label,
)
outer_loop = state.outer_loop
libs/partners/quickjs/langchain_quickjs/_subagent.py
+193 -13
View File
@@ -3,9 +3,11 @@
from __future__ import annotations
import json
import logging
import time
import uuid
from dataclasses import replace
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Final, Literal, NotRequired, TypedDict
from langchain.agents.structured_output import AutoStrategy
@@ -22,10 +24,141 @@ if TYPE_CHECKING:
from langchain_core.tools import BaseTool
logger = logging.getLogger(__name__)
SUBAGENT_STREAM_EVENT_TYPE: Final = "subagent"
"""Discriminator value for subagent events on the custom stream."""
_SCHEMA_MAX_BYTES = 4096
"""Maximum serialized size of an accepted `response_schema`."""
_SCHEMA_MAX_DEPTH = 5
"""Maximum nesting depth allowed in a `response_schema`."""
_SCHEMA_MAX_PROPERTIES = 32
"""Maximum total property count allowed across a `response_schema`."""
_SUBAGENT_TASK_TOOL_FIELDS = frozenset({"description", "subagent_type"})
"""Input field names that identify the Deep Agents task tool."""
_EVENT_DESCRIPTION_MAX_CHARS = 200
"""Character cap on the `description` carried in a start event."""
_EVENT_LABEL_MAX_CHARS = 120
"""Character cap on an explicit `label` carried in a start event."""
_EVENT_LABEL_FALLBACK_MAX_CHARS = 60
"""Character cap on a label derived from the description fallback."""
class SubagentStartEvent(TypedDict):
"""A subagent began running inside a `js_eval` call."""
id: str
"""Per-dispatch id, stable across this subagent's start/complete/error."""
type: Literal["subagent"]
"""Stream event discriminator; always `subagent`."""
phase: Literal["start"]
"""Lifecycle phase for this event."""
eval_id: NotRequired[str]
"""Parent `js_eval` tool-call id, used to group a fan-out by batch.
Omitted when the runtime exposes no `tool_call_id`.
"""
subagent_type: str
"""The dispatched subagent type (the `subagentType` argument)."""
label: str
"""Short row label; falls back to a compact description when unset."""
description: str
"""The task description, truncated for display."""
class SubagentCompleteEvent(TypedDict):
"""A subagent finished successfully inside a `js_eval` call."""
id: str
"""Per-dispatch id, matching the corresponding `start` event."""
type: Literal["subagent"]
"""Stream event discriminator; always `subagent`."""
phase: Literal["complete"]
"""Lifecycle phase for this event."""
eval_id: NotRequired[str]
"""Parent `js_eval` tool-call id; omitted when the runtime exposes none."""
duration_ms: int
"""Wall-clock duration of the subagent, in milliseconds."""
class SubagentErrorEvent(TypedDict):
"""A subagent raised before returning inside a `js_eval` call."""
id: str
"""Per-dispatch id, matching the corresponding `start` event."""
type: Literal["subagent"]
"""Stream event discriminator; always `subagent`."""
phase: Literal["error"]
"""Lifecycle phase for this event."""
eval_id: NotRequired[str]
"""Parent `js_eval` tool-call id; omitted when the runtime exposes none."""
duration_ms: int
"""Wall-clock duration before the failure, in milliseconds."""
error: str
"""The failure string (`str(exc)` of the raised exception)."""
SubagentStreamEvent = SubagentStartEvent | SubagentCompleteEvent | SubagentErrorEvent
"""One lifecycle event for a subagent dispatched from inside `js_eval`.
Emitted on LangGraph's `custom` stream so UIs can render a live fan-out panel.
A `phase`-discriminated union: `start` carries the descriptive fields,
`complete`/`error` carry the measured `duration_ms`, and `error` carries the
failure string. `type`/`phase`/`id` are always present.
Consumers should tolerate unrecognized `phase` values rather than assume the
union is closed, so a future phase can be added without breaking them.
"""
def _emit_subagent_event(stream_writer: Any, event: SubagentStreamEvent) -> None:
"""Emit a subagent lifecycle event on the custom stream.
Any failure is swallowed so observability never breaks dispatch.
"""
if stream_writer is None:
return
try:
stream_writer(event)
except Exception: # noqa: BLE001 — observability must not break dispatch
# Use `.get` rather than subscripting: this handler must never raise,
# regardless of how well-formed the event that reached it was.
logger.debug(
"Failed to emit subagent stream event (id=%s, phase=%s)",
event.get("id"),
event.get("phase"),
exc_info=True,
)
def _event_label(label: str | None, description: str) -> str:
"""Return the explicit label or a compact description fallback."""
explicit = " ".join(label.split()) if label else ""
if explicit:
return explicit[:_EVENT_LABEL_MAX_CHARS]
return " ".join(description.split())[:_EVENT_LABEL_FALLBACK_MAX_CHARS]
def find_subagent_task_tool(tools: Sequence[BaseTool]) -> BaseTool | None:
@@ -61,8 +194,13 @@ async def call_subagent_task_tool(
subagent_type: str,
response_schema: dict[str, Any] | None,
runtime: Any,
label: str | None = None,
) -> Any:
"""Call the Deep Agents task tool and return a JavaScript-friendly value."""
"""Call the Deep Agents task tool and return a JavaScript-friendly value.
This also emits `start` then `complete`/`error` subagent lifecycle
events on the custom stream.
"""
if runtime is None:
msg = "task() requires an active ToolRuntime"
raise RuntimeError(msg)
@@ -73,18 +211,60 @@ async def call_subagent_task_tool(
response_schema = _ensure_schema_title(response_schema)
runtime = _runtime_with_response_format(runtime, response_schema)
runtime = _runtime_with_tool_call_id(
runtime,
f"ptc_{task_tool.name}_{uuid.uuid4().hex[:8]}",
)
result = await task_tool.arun(
{
"description": description,
"subagent_type": subagent_type,
"runtime": runtime,
eval_id = getattr(runtime, "tool_call_id", None)
stream_writer = getattr(runtime, "stream_writer", None)
subagent_id = f"ptc_{task_tool.name}_{uuid.uuid4().hex[:8]}"
runtime = _runtime_with_tool_call_id(runtime, subagent_id)
start_event: SubagentStartEvent = {
"type": SUBAGENT_STREAM_EVENT_TYPE,
"phase": "start",
"id": subagent_id,
"subagent_type": subagent_type,
"label": _event_label(label, description),
"description": description[:_EVENT_DESCRIPTION_MAX_CHARS],
}
# Only carry `eval_id` when the runtime exposes a parent tool-call id;
# omitting it (rather than sending None) keeps the wire type tight and lets
# consumers distinguish "no parent batch" from a real id.
if eval_id is not None:
start_event["eval_id"] = eval_id
_emit_subagent_event(stream_writer, start_event)
started_at = time.monotonic()
try:
result = await task_tool.arun(
{
"description": description,
"subagent_type": subagent_type,
"runtime": runtime,
}
)
except Exception as e:
error_event: SubagentErrorEvent = {
"type": SUBAGENT_STREAM_EVENT_TYPE,
"phase": "error",
"id": subagent_id,
"duration_ms": int((time.monotonic() - started_at) * 1000),
"error": str(e),
}
)
return _extract_task_tool_output(result, parse_json_output=parse_json_output)
if eval_id is not None:
error_event["eval_id"] = eval_id
_emit_subagent_event(stream_writer, error_event)
raise
output = _extract_task_tool_output(result, parse_json_output=parse_json_output)
complete_event: SubagentCompleteEvent = {
"type": SUBAGENT_STREAM_EVENT_TYPE,
"phase": "complete",
"id": subagent_id,
"duration_ms": int((time.monotonic() - started_at) * 1000),
}
if eval_id is not None:
complete_event["eval_id"] = eval_id
_emit_subagent_event(stream_writer, complete_event)
return output
def _validate_response_schema(schema: dict[str, Any]) -> None:
libs/partners/quickjs/tests/unit_tests/smoke_tests/snapshots/quickjs_system_prompt_mixed_foreign_functions.md
+9 -2
View File
@@ -143,6 +143,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -160,6 +161,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -334,8 +339,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/tests/unit_tests/smoke_tests/snapshots/quickjs_system_prompt_mixed_foreign_functions_call.md
+9 -2
View File
@@ -143,6 +143,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -160,6 +161,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -334,8 +339,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/tests/unit_tests/smoke_tests/snapshots/quickjs_system_prompt_mixed_foreign_functions_turn.md
+9 -2
View File
@@ -143,6 +143,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -160,6 +161,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -334,8 +339,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/tests/unit_tests/smoke_tests/snapshots/quickjs_system_prompt_no_tools.md
+9 -2
View File
@@ -143,6 +143,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -160,6 +161,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -334,8 +339,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/tests/unit_tests/smoke_tests/snapshots/quickjs_system_prompt_no_tools_call.md
+9 -2
View File
@@ -143,6 +143,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -160,6 +161,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -334,8 +339,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/tests/unit_tests/smoke_tests/snapshots/quickjs_system_prompt_no_tools_turn.md
+9 -2
View File
@@ -143,6 +143,7 @@ flow, and synthesis - in plain JavaScript.
await task({
description, // full autonomous task prompt
subagentType, // configured subagent name
label, // optional short UI label for this dispatch
responseSchema, // optional JSON Schema for structured output
}); // -> Promise<unknown>
```
@@ -160,6 +161,10 @@ exploring, still pass its path and let the subagent read it; do not paste back
what you read. Each dispatch is stateless from the caller's perspective; you
cannot send follow-up messages to the same subagent run.
`label` is optional: when provided, it is shown in the live progress UI
instead of the default description-derived fallback. It is not sent to the
subagent and does not affect execution.
`responseSchema` is optional, but set it on any dispatch whose result feeds
later code. A deterministic, typed shape is what lets you compose the next
stage reliably — index it, sort it, compare fields, branch on it, merge it —
@@ -334,8 +339,10 @@ const findings = auditResults.flatMap((r) =>
r.findings.map((f) => ({ ...f, file: r.file }))
);
const verified = await Promise.all(findings.map((f) =>
task({ description: "Verify this finding: " + f.evidence, subagentType: "verifier" })
.then((v) => ({ ...f, ...v }))
task({
description: "Verify this finding: " + f.evidence,
subagentType: "verifier",
}).then((v) => ({ ...f, ...v }))
));
```
libs/partners/quickjs/tests/unit_tests/test_repl_middleware.py
+18 -6
View File
@@ -1077,17 +1077,22 @@ async def test_async_task_global_not_installed_when_disabled(
("code", "message"),
[
(
"await task({subagentType: 'worker'})",
"await task({subagentType: 'worker', label: 'lbl'})",
"task() requires non-empty string field `description`",
),
(
"await task({description: 'work'})",
"await task({description: 'work', label: 'lbl'})",
"task() requires non-empty string field `subagentType`",
),
(
"await task({description: 'work', subagentType: 'worker', label: 123})",
"task() field `label` must be a string when provided",
),
(
"await task({"
"description: 'work', "
"subagentType: 'worker', "
"label: 'lbl', "
"responseSchema: 'bad'"
"})",
"task() field `responseSchema` must be an object when provided",
@@ -1124,7 +1129,7 @@ async def test_async_task_global_missing_task_tool_surfaces_as_eval_error(
)
outcome = await repl.eval_async(
"await task({description: 'work', subagentType: 'worker'})",
"await task({description: 'work', subagentType: 'worker', label: 'lbl'})",
outer_runtime=runtime,
)
@@ -1144,11 +1149,13 @@ async def test_async_task_global_returns_declarative_structured_response_object(
"const first = await task({"
"description: 'work', "
"subagentType: 'worker', "
"label: 'first', "
"responseSchema: schema"
"});"
"const second = await task({"
"description: 'work again', "
"subagentType: 'worker', "
"label: 'second', "
"responseSchema: schema"
"});"
"JSON.stringify({"
@@ -1177,6 +1184,7 @@ async def test_async_task_global_rejects_compiled_response_schema(
"await task({"
"description: 'work', "
"subagentType: 'worker', "
"label: 'lbl', "
"responseSchema: {"
"type: 'object', "
"properties: {ok: {type: 'boolean'}}, "
@@ -1206,7 +1214,9 @@ async def test_async_task_global_uses_last_non_empty_ai_message(
)
outcome = await repl.eval_async(
"JSON.stringify(await task({description: 'work', subagentType: 'worker'}))",
"JSON.stringify(await task({"
"description: 'work', subagentType: 'worker', label: 'lbl'"
"}))",
outer_runtime=_subagent_runtime(runnable),
)
@@ -1227,7 +1237,7 @@ async def test_async_task_global_rejects_state_without_messages(
)
outcome = await repl.eval_async(
"await task({description: 'work', subagentType: 'worker'})",
"await task({description: 'work', subagentType: 'worker', label: 'lbl'})",
outer_runtime=_subagent_runtime(runnable),
)
@@ -1279,7 +1289,9 @@ async def test_async_task_global_limits_concurrency_per_repl(
outcome = await repl.eval_async(
"const calls = [];"
"for (let i = 0; i < 64; i++) {"
" calls.push(task({description: String(i), subagentType: 'worker'}));"
" calls.push(task({"
"description: String(i), subagentType: 'worker', label: 'c' + i"
"}));"
"}"
"(await Promise.all(calls)).length",
outer_runtime=_subagent_runtime(runnable),
libs/partners/quickjs/tests/unit_tests/test_subagent_events.py
+229
View File
@@ -0,0 +1,229 @@
"""Tests for live subagent lifecycle events emitted on the custom stream.
`call_subagent_task_tool` emits start/complete (or error) events via the
runtime's `stream_writer` so a UI can render a live fan-out panel. These tests
cover event shape, ordering, id propagation, truncation, and that telemetry
failures never break the underlying dispatch.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
import pytest
from langchain_quickjs._subagent import call_subagent_task_tool
@dataclass
class _FakeRuntime:
"""Minimal stand-in for the LangGraph ToolRuntime the bridge passes in."""
tool_call_id: str = "eval_call_123"
stream_writer: Any = None
config: dict | None = None
class _FakeTaskTool:
"""Stand-in for the deepagents `task` tool."""
name = "task"
def __init__(
self,
result: str = "done",
*,
raise_exc: Exception | None = None,
) -> None:
self._result = result
self._raise = raise_exc
self.seen_runtime_tool_call_id: str | None = None
async def arun(self, args: dict[str, Any], **_kwargs: Any) -> str:
self.seen_runtime_tool_call_id = args["runtime"].tool_call_id
if self._raise is not None:
raise self._raise
return self._result
@dataclass
class _Recorder:
events: list[dict[str, Any]] = field(default_factory=list)
def __call__(self, event: dict[str, Any]) -> None:
self.events.append(event)
async def test_emits_start_then_complete() -> None:
rec = _Recorder()
runtime = _FakeRuntime(stream_writer=rec)
tool = _FakeTaskTool("hello world")
out = await call_subagent_task_tool(
tool,
description="do the thing",
subagent_type="researcher",
label="lbl",
response_schema=None,
runtime=runtime,
)
assert out == "hello world"
assert [e["phase"] for e in rec.events] == ["start", "complete"]
start, complete = rec.events
assert start["type"] == "subagent"
assert start["eval_id"] == "eval_call_123"
assert start["subagent_type"] == "researcher"
assert start["label"] == "lbl"
assert start["description"] == "do the thing"
# The per-dispatch id is stable across start/complete and is the fresh
# child tool_call_id (not the parent eval id).
assert start["id"] == complete["id"]
assert start["id"].startswith("ptc_task_")
assert tool.seen_runtime_tool_call_id == start["id"]
assert isinstance(complete["duration_ms"], int)
async def test_emits_error_event_and_reraises() -> None:
rec = _Recorder()
runtime = _FakeRuntime(stream_writer=rec)
tool = _FakeTaskTool(raise_exc=ValueError("boom"))
with pytest.raises(ValueError, match="boom"):
await call_subagent_task_tool(
tool,
description="x",
subagent_type="t",
label="lbl",
response_schema=None,
runtime=runtime,
)
assert [e["phase"] for e in rec.events] == ["start", "error"]
error = rec.events[1]
assert error["error"] == "boom"
assert error["id"] == rec.events[0]["id"]
assert isinstance(error["duration_ms"], int)
async def test_description_is_truncated_in_event() -> None:
rec = _Recorder()
runtime = _FakeRuntime(stream_writer=rec)
await call_subagent_task_tool(
_FakeTaskTool("r"),
description="a" * 500,
subagent_type="t",
label="lbl",
response_schema=None,
runtime=runtime,
)
assert len(rec.events[0]["description"]) == 200
async def test_missing_label_falls_back_to_short_description() -> None:
rec = _Recorder()
runtime = _FakeRuntime(stream_writer=rec)
description = "Review\n\n" + "a" * 100
await call_subagent_task_tool(
_FakeTaskTool("r"),
description=description,
subagent_type="t",
response_schema=None,
runtime=runtime,
)
assert rec.events[0]["label"] == ("Review " + "a" * 100)[:60]
async def test_label_is_truncated_in_event() -> None:
rec = _Recorder()
runtime = _FakeRuntime(stream_writer=rec)
await call_subagent_task_tool(
_FakeTaskTool("r"),
description="x",
subagent_type="t",
label="L" * 500,
response_schema=None,
runtime=runtime,
)
assert len(rec.events[0]["label"]) == 120
async def test_writer_failure_does_not_break_dispatch() -> None:
def boom_writer(_event: dict[str, Any]) -> None:
msg = "writer down"
raise RuntimeError(msg)
runtime = _FakeRuntime(stream_writer=boom_writer)
out = await call_subagent_task_tool(
_FakeTaskTool("still works"),
description="x",
subagent_type="t",
label="lbl",
response_schema=None,
runtime=runtime,
)
assert out == "still works"
async def test_missing_writer_is_a_noop() -> None:
runtime = _FakeRuntime(stream_writer=None)
out = await call_subagent_task_tool(
_FakeTaskTool("ok"),
description="x",
subagent_type="t",
label="lbl",
response_schema=None,
runtime=runtime,
)
assert out == "ok"
async def test_missing_tool_call_id_omits_eval_id() -> None:
# When the runtime exposes no tool_call_id, `eval_id` is omitted from the
# wire event entirely (rather than sent as None) so consumers can tell
# "no parent batch" from a real id.
rec = _Recorder()
runtime = _FakeRuntime(tool_call_id=None, stream_writer=rec)
await call_subagent_task_tool(
_FakeTaskTool("r"),
description="x",
subagent_type="t",
label="lbl",
response_schema=None,
runtime=runtime,
)
assert [e["phase"] for e in rec.events] == ["start", "complete"]
for event in rec.events:
assert "eval_id" not in event
async def test_structured_output_path_still_emits_events() -> None:
# Setting response_schema replaces the runtime and changes output parsing;
# the start/complete lifecycle events must still fire around it.
rec = _Recorder()
runtime = _FakeRuntime(stream_writer=rec)
out = await call_subagent_task_tool(
_FakeTaskTool('{"answer": 42}'),
description="x",
subagent_type="t",
label="lbl",
response_schema={
"type": "object",
"properties": {"answer": {"type": "number"}},
},
runtime=runtime,
)
assert out == {"answer": 42}
assert [e["phase"] for e in rec.events] == ["start", "complete"]
assert rec.events[0]["eval_id"] == "eval_call_123"
libs/partners/quickjs/tests/unit_tests/test_thread_affinity.py
+4 -2
View File
@@ -131,7 +131,8 @@ async def test_quickjs_async_task_global_subagent_loop_affinity_e2e() -> None:
agent = create_deep_agent(
model=_FakeChatModel(
messages=_script(
"await task({description: 'say hi', subagentType: 'researcher'})",
"await task({description: 'say hi', "
"subagentType: 'researcher', label: 'greet'})",
final_message="done",
)
),
@@ -172,7 +173,8 @@ async def test_quickjs_task_global_available_without_ptc_e2e() -> None:
agent = create_deep_agent(
model=_FakeChatModel(
messages=_script(
"await task({description: 'say hi', subagentType: 'researcher'})",
"await task({description: 'say hi', "
"subagentType: 'researcher', label: 'greet'})",
final_message="done",
)
),