feat(agent): read_window_below tool — which OS window is underneath the desktop app

Desktop-gated (desktop_ui toolset) metadata-only window awareness: the agent
can ask which application window sits directly behind the Hermes window
(app, title, bounds — never pixels). Rides the same blocking bridge as
read_terminal: the gateway emits window.read.request and the renderer
answers window.read.respond.
This commit is contained in:
Brooklyn Nicholson 2026-08-07 22:27:28 -05:00 committed by brooklyn!
parent a6ede70c2a
commit 406501fd97
9 changed files with 171 additions and 1 deletions

View File

@ -493,6 +493,7 @@ def init_agent(
clarify_callback: callable = None,
read_terminal_callback: callable = None,
read_preview_callback: callable = None,
read_window_below_callback: callable = None,
step_callback: callable = None,
stream_delta_callback: callable = None,
interim_assistant_callback: callable = None,
@ -767,6 +768,7 @@ def init_agent(
agent.clarify_callback = clarify_callback
agent.read_terminal_callback = read_terminal_callback
agent.read_preview_callback = read_preview_callback
agent.read_window_below_callback = read_window_below_callback
agent.step_callback = step_callback
agent.stream_delta_callback = stream_delta_callback
agent.interim_assistant_callback = interim_assistant_callback

View File

@ -99,7 +99,7 @@ def _ra():
AGENT_RUNTIME_POST_HOOK_TOOL_NAMES = frozenset(
{"todo", "session_search", "memory", "clarify", "read_terminal", "read_preview", "delegate_task"}
{"todo", "session_search", "memory", "clarify", "read_terminal", "read_preview", "read_window_below", "delegate_task"}
)
@ -3000,6 +3000,15 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
),
next_args,
)
elif function_name == "read_window_below":
def _execute(next_args: dict) -> Any:
from tools.read_window_tool import read_window_below_tool as _read_window_below_tool
return _finish_agent_tool(
_read_window_below_tool(
callback=getattr(agent, "read_window_below_callback", None),
),
next_args,
)
elif function_name == "delegate_task":
def _execute(next_args: dict) -> Any:
return _finish_agent_tool(agent._dispatch_delegate_task(next_args), next_args)

View File

@ -1880,6 +1880,25 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
tool_duration = time.time() - tool_start_time
if agent._should_emit_quiet_tool_messages():
agent._vprint(f" {_get_cute_tool_message_impl('read_preview', function_args, tool_duration, result=function_result)}")
elif function_name == "read_window_below":
def _execute(next_args: dict) -> Any:
from tools.read_window_tool import read_window_below_tool as _read_window_below_tool
return _read_window_below_tool(
callback=getattr(agent, "read_window_below_callback", None),
)
function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware(
agent,
function_name=function_name,
function_args=function_args,
effective_task_id=effective_task_id,
tool_call_id=getattr(tool_call, "id", "") or "",
execute=_execute,
scope_block=_ts_scope_block,
display_index=i,
))
tool_duration = time.time() - tool_start_time
if agent._should_emit_quiet_tool_messages():
agent._vprint(f" {_get_cute_tool_message_impl('read_window_below', function_args, tool_duration, result=function_result)}")
elif function_name == "delegate_task":
tasks_arg = function_args.get("tasks")
if tasks_arg and isinstance(tasks_arg, list):

View File

@ -470,6 +470,7 @@ class AIAgent:
clarify_callback: callable = None,
read_terminal_callback: callable = None,
read_preview_callback: callable = None,
read_window_below_callback: callable = None,
step_callback: callable = None,
stream_delta_callback: callable = None,
interim_assistant_callback: callable = None,
@ -556,6 +557,7 @@ class AIAgent:
clarify_callback=clarify_callback,
read_terminal_callback=read_terminal_callback,
read_preview_callback=read_preview_callback,
read_window_below_callback=read_window_below_callback,
step_callback=step_callback,
stream_delta_callback=stream_delta_callback,
interim_assistant_callback=interim_assistant_callback,

View File

@ -0,0 +1,50 @@
"""Tests for the GUI-surface ``read_window_below`` tool."""
import json
from tools import read_window_tool as rw
from tools.registry import registry
def test_lives_in_the_gui_surface_toolset(monkeypatch):
"""Mirrors read_terminal: scoped by toolset, not by the backend's env."""
monkeypatch.delenv("HERMES_DESKTOP", raising=False)
entry = registry.get_entry("read_window_below")
assert entry is not None
assert entry.toolset == "desktop_ui"
assert entry.check_fn is None
def test_requires_callback():
"""Outside the desktop GUI there is no bridge — a clear error, no crash."""
result = json.loads(rw.read_window_below_tool(callback=None))
assert "desktop" in result["error"]
def test_empty_answer_means_unavailable():
result = json.loads(rw.read_window_below_tool(callback=lambda: ""))
assert "error" in result
def test_passes_json_through():
payload = {
"window": {"app": "Figma", "title": "", "bounds": {"x": 0, "y": 38, "width": 1470, "height": 870}, "id": 13937},
"frontmost": {"app": "Figma", "title": ""},
"platform": "darwin",
}
result = json.loads(rw.read_window_below_tool(callback=lambda: json.dumps(payload)))
assert result == payload
def test_wraps_non_json_text():
result = json.loads(rw.read_window_below_tool(callback=lambda: "plain words"))
assert result == {"text": "plain words"}
def test_callback_failure_is_reported():
def _boom():
raise RuntimeError("renderer went away")
result = json.loads(rw.read_window_below_tool(callback=_boom))
assert "renderer went away" in result["error"]

67
tools/read_window_tool.py Normal file
View File

@ -0,0 +1,67 @@
#!/usr/bin/env python3
"""Read which OS window sits directly underneath the Hermes desktop window.
The window list lives with the OS, so this tool round-trips through the
gateway's blocking-prompt bridge — the same one `read_terminal` uses:
tui_gateway emits ``window.read.request``, the desktop renderer asks its main
process (which owns native window enumeration) and answers with
``window.read.respond``. This module is just schema + a thin dispatcher over
the platform-injected callback.
"""
import json
from typing import Callable, Optional
from tools.registry import registry, tool_error
def read_window_below_tool(callback: Optional[Callable] = None) -> str:
"""Return the window underneath the Hermes window as a JSON string."""
if callback is None:
return tool_error(
"read_window_below is only available in the Hermes desktop app."
)
try:
raw = callback()
except Exception as exc:
return tool_error(f"Failed to read the window below: {exc}")
if not raw:
return tool_error(
"Could not determine the window underneath (the desktop app did "
"not answer, or window enumeration is unavailable on this system)."
)
# Desktop answers with a JSON object; pass it through, else wrap the raw text.
try:
return json.dumps(json.loads(raw), ensure_ascii=False)
except (TypeError, ValueError):
return json.dumps({"text": str(raw)}, ensure_ascii=False)
READ_WINDOW_BELOW_SCHEMA = {
"name": "read_window_below",
"description": (
"Identify the application window directly underneath (behind) the "
"Hermes desktop window — what the user is working in behind this app. "
"Returns JSON: {window: {app, title, bounds{x,y,width,height}, id}, "
"frontmost: {app, title}, platform}. `title` may be empty when the OS "
"withholds window titles (e.g. macOS without the Screen Recording "
"permission — never prompted for, noted in `note`). Metadata only; "
"this never captures pixels or content of other windows."
),
"parameters": {
"type": "object",
"properties": {},
},
}
registry.register(
name="read_window_below",
toolset="desktop_ui",
schema=READ_WINDOW_BELOW_SCHEMA,
handler=lambda args, **kw: read_window_below_tool(callback=kw.get("callback")),
emoji="🪟",
)

View File

@ -273,6 +273,7 @@ TOOLSETS = {
"tools": [
"read_terminal", "close_terminal",
"open_preview", "read_preview",
"read_window_below",
"focus_pane", "react_to_message",
],
"includes": []

View File

@ -936,6 +936,15 @@ def _(rid, params: dict) -> dict:
return _respond(rid, params, "text", allow_expired=True)
@method("window.read.respond")
def _(rid, params: dict) -> dict:
# `text` is a JSON string describing the OS window underneath the Hermes
# window (read_window_below tool). allow_expired=True for the same reason
# as terminal.read: the tool's bounded wait can expire while the renderer's
# round-trip to the main process is still in flight.
return _respond(rid, params, "text", allow_expired=True)
@method("sudo.respond")
def _(rid, params: dict) -> dict:
return _respond(rid, params, "password", allow_expired=True)

View File

@ -3240,6 +3240,7 @@ def _block(event: str, sid: str, payload: dict, timeout: float | None = 300) ->
"clarify.request",
"terminal.read.request",
"preview.read.request",
"window.read.request",
}:
_emit(
f"{event.removesuffix('.request')}.expire",
@ -5809,6 +5810,16 @@ def _agent_cbs(sid: str) -> dict:
{k: v for k, v in (("start", start), ("count", count)) if v is not None},
timeout=45,
),
# read_window_below tool (desktop GUI): the renderer asks its main
# process (which owns native window enumeration) which OS window sits
# directly underneath the Hermes window, and answers
# window.read.respond with the serialized metadata.
"read_window_below_callback": lambda: _block(
"window.read.request",
sid,
{},
timeout=30,
),
}
# Interim assistant commentary (text alongside tool calls, or the attempted