Skip to content

feat(linux): port type_into_editable from pyatspi subprocess to native AT-SPI #1853

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
r33drichards wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(linux): port type_into_editable from pyatspi subprocess to native AT-SPI #1853

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 2 additions & 56 deletions libs/cua-driver/rust/crates/platform-linux/src/atspi/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ pub struct AtspiNode {
pub value: Option<String>,
pub description: Option<String>,
pub actions: Vec<String>,
/// For pyatspi path: element_key = element_index as u64.
/// For AT-SPI path: element_key = element_index as u64.
/// For X11 fallback: element_key = xid.
pub element_key: u64,
}
Expand Down Expand Up @@ -59,61 +59,7 @@ pub fn perform_action(pid: u32, idx: usize) -> Result<String> {
/// For Qt5, which doesn't expose widgets when unfocused, this will return Err.
/// Returns Ok if an editable was found and text was set, Err otherwise.
pub fn type_into_editable(pid: u32, text: &str) -> Result<()> {
let safe_text = text.replace('\\', "\\\\").replace('\'', "\\'");
let script = format!(r#"
import pyatspi, sys

def find_editable(acc, depth=0):
# Try to find any EditableText interface, regardless of role
try:
et = acc.queryEditableText()
# If we can query it, return this node
return acc
except:
pass

# Recursively search children
try:
for child in acc:
result = find_editable(child, depth + 1)
if result is not None:
return result
except:
pass

return None

desktop = pyatspi.Registry.getDesktop(0)
editable = None
for app in desktop:
try:
if app.get_process_id() == {pid}:
for win in app:
editable = find_editable(win)
if editable:
break
break
except:
pass

if editable is None:
print("ERROR: No editable found", file=sys.stderr)
sys.exit(1)

try:
et = editable.queryEditableText()
et.setTextContents('{safe_text}')
print("ok:atspi")
except Exception as e:
print(f"ERROR: {{e}}", file=sys.stderr)
sys.exit(1)
"#, pid = pid, safe_text = safe_text);

let out = std::process::Command::new("python3").arg("-c").arg(&script).output()?;
if !out.status.success() {
anyhow::bail!("{}", String::from_utf8_lossy(&out.stderr).trim().to_owned());
}
Ok(())
native::type_into_editable(pid, text)
}

/// Set the text value of element `idx` within pid's app tree via AT-SPI.
Expand Down
37 changes: 37 additions & 0 deletions libs/cua-driver/rust/crates/platform-linux/src/atspi/native.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -540,6 +540,43 @@ pub fn insert_text(pid: u32, text: &str) -> Result<bool> {
})
}

/// Replace the contents of the best editable field in pid's tree with `text`
/// via AT-SPI EditableText (setTextContents semantics, unlike [`insert_text`]
/// which inserts at the caret). Works on unfocused windows when the toolkit
/// exposes EditableText (Qt6, GTK4); errs when no editable is exposed (Qt5
/// unfocused) so the caller can run its focus workaround and retry.
pub fn type_into_editable(pid: u32, text: &str) -> Result<()> {
runtime().block_on(async {
let conn = AccessibilityConnection::new()
.await
.map_err(|e| anyhow!("AT-SPI connect failed: {e}"))?;
let visited = collect_visited(&conn, pid)
.await?
.ok_or_else(|| anyhow!("no AT-SPI application for pid {pid}"))?;
let target = pick_editable(&visited)
.ok_or_else(|| anyhow!("no editable element found for pid {pid}"))?;
dlog!(
"type_into_editable target: role={:?} focused={} in_web_doc={}",
target.role, target.focused, target.in_web_doc
);

let et = target
.acc
.proxies()
.await
.map_err(|e| anyhow!("interface proxies unavailable: {e}"))?
.editable_text()
.await
.map_err(|e| anyhow!("EditableText unavailable: {e}"))?;
match call(et.set_text_contents(text)).await {
Some(Ok(true)) => Ok(()),
Some(Ok(false)) => Err(anyhow!("setTextContents rejected by toolkit")),
Some(Err(e)) => Err(anyhow!("setTextContents failed: {e}")),
None => Err(anyhow!("setTextContents timed out")),
}
})
}
Comment on lines +549 to +578

@coderabbitai coderabbitai Bot Jun 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Add an overall timeout for type_into_editable to keep latency bounded.

Line 549 executes the full tree walk/write path without OP_TIMEOUT; on large/unresponsive trees this can block for a long time and tie up blocking workers.

⏱️ Suggested fix 
 pub fn type_into_editable(pid: u32, text: &str) -> Result<()> {
     runtime().block_on(async {
-        let conn = AccessibilityConnection::new()
-            .await
-            .map_err(|e| anyhow!("AT-SPI connect failed: {e}"))?;
-        let visited = collect_visited(&conn, pid)
-            .await?
-            .ok_or_else(|| anyhow!("no AT-SPI application for pid {pid}"))?;
-        let target = pick_editable(&visited)
-            .ok_or_else(|| anyhow!("no editable element found for pid {pid}"))?;
-        dlog!(
-            "type_into_editable target: role={:?} focused={} in_web_doc={}",
-            target.role, target.focused, target.in_web_doc
-        );
-
-        let et = target
-            .acc
-            .proxies()
-            .await
-            .map_err(|e| anyhow!("interface proxies unavailable: {e}"))?
-            .editable_text()
-            .await
-            .map_err(|e| anyhow!("EditableText unavailable: {e}"))?;
-        match call(et.set_text_contents(text)).await {
-            Some(Ok(true)) => Ok(()),
-            Some(Ok(false)) => Err(anyhow!("setTextContents rejected by toolkit")),
-            Some(Err(e)) => Err(anyhow!("setTextContents failed: {e}")),
-            None => Err(anyhow!("setTextContents timed out")),
+        let work = async {
+            let conn = AccessibilityConnection::new()
+                .await
+                .map_err(|e| anyhow!("AT-SPI connect failed: {e}"))?;
+            let visited = collect_visited(&conn, pid)
+                .await?
+                .ok_or_else(|| anyhow!("no AT-SPI application for pid {pid}"))?;
+            let target = pick_editable(&visited)
+                .ok_or_else(|| anyhow!("no editable element found for pid {pid}"))?;
+            dlog!(
+                "type_into_editable target: role={:?} focused={} in_web_doc={}",
+                target.role, target.focused, target.in_web_doc
+            );
+
+            let et = target
+                .acc
+                .proxies()
+                .await
+                .map_err(|e| anyhow!("interface proxies unavailable: {e}"))?
+                .editable_text()
+                .await
+                .map_err(|e| anyhow!("EditableText unavailable: {e}"))?;
+            match call(et.set_text_contents(text)).await {
+                Some(Ok(true)) => Ok(()),
+                Some(Ok(false)) => Err(anyhow!("setTextContents rejected by toolkit")),
+                Some(Err(e)) => Err(anyhow!("setTextContents failed: {e}")),
+                None => Err(anyhow!("setTextContents timed out")),
+            }
+        };
+        match tokio::time::timeout(OP_TIMEOUT, work).await {
+            Ok(r) => r,
+            Err(_) => Err(anyhow!("type_into_editable timed out for pid {pid}")),
         }
     })
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
runtime().block_on(async {
let conn = AccessibilityConnection::new()
.await
.map_err(|e| anyhow!("AT-SPI connect failed: {e}"))?;
let visited = collect_visited(&conn, pid)
.await?
.ok_or_else(|| anyhow!("no AT-SPI application for pid {pid}"))?;
let target = pick_editable(&visited)
.ok_or_else(|| anyhow!("no editable element found for pid {pid}"))?;
dlog!(
"type_into_editable target: role={:?} focused={} in_web_doc={}",
target.role, target.focused, target.in_web_doc
);
let et = target
.acc
.proxies()
.await
.map_err(|e| anyhow!("interface proxies unavailable: {e}"))?
.editable_text()
.await
.map_err(|e| anyhow!("EditableText unavailable: {e}"))?;
match call(et.set_text_contents(text)).await {
Some(Ok(true)) => Ok(()),
Some(Ok(false)) => Err(anyhow!("setTextContents rejected by toolkit")),
Some(Err(e)) => Err(anyhow!("setTextContents failed: {e}")),
None => Err(anyhow!("setTextContents timed out")),
}
})
}
pub fn type_into_editable(pid: u32, text: &str) -> Result<()> {
runtime().block_on(async {
let work = async {
let conn = AccessibilityConnection::new()
.await
.map_err(|e| anyhow!("AT-SPI connect failed: {e}"))?;
let visited = collect_visited(&conn, pid)
.await?
.ok_or_else(|| anyhow!("no AT-SPI application for pid {pid}"))?;
let target = pick_editable(&visited)
.ok_or_else(|| anyhow!("no editable element found for pid {pid}"))?;
dlog!(
"type_into_editable target: role={:?} focused={} in_web_doc={}",
target.role, target.focused, target.in_web_doc
);
let et = target
.acc
.proxies()
.await
.map_err(|e| anyhow!("interface proxies unavailable: {e}"))?
.editable_text()
.await
.map_err(|e| anyhow!("EditableText unavailable: {e}"))?;
match call(et.set_text_contents(text)).await {
Some(Ok(true)) => Ok(()),
Some(Ok(false)) => Err(anyhow!("setTextContents rejected by toolkit")),
Some(Err(e)) => Err(anyhow!("setTextContents failed: {e}")),
None => Err(anyhow!("setTextContents timed out")),
}
};
match tokio::time::timeout(OP_TIMEOUT, work).await {
Ok(r) => r,
Err(_) => Err(anyhow!("type_into_editable timed out for pid {pid}")),
}
})
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@libs/cua-driver/rust/crates/platform-linux/src/atspi/native.rs` around lines
549 - 578, The async work in type_into_editable (the runtime().block_on(async {
... }) block that calls AccessibilityConnection::new, collect_visited,
pick_editable, obtains editable_text and calls set_text_contents via
call(...).await) must be bounded by an overall timeout; wrap that entire async
operation with tokio::time::timeout (or equivalent OP_TIMEOUT) and convert a
timeout result into an anyhow error like "type_into_editable timed out" so
callers receive a clear timeout error instead of a hung blocking worker; ensure
existing per-step map_err branches remain but return a timeout error when
tokio::time::timeout returns Err.


/// Find the window XID for a PID by listing its X11 windows.
async fn entry_find_window_xid(pid: u32) -> Option<u64> {
use crate::x11::list_windows;
Expand Down
2 changes: 1 addition & 1 deletion libs/cua-driver/rust/crates/platform-linux/src/tools/impl_.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,7 +520,7 @@ impl Tool for LaunchAppTool {
/// Resolve an AT-SPI element's center in window-local X11 coordinates.
///
/// Returns `(xid, window_local_x, window_local_y)`.
/// Looks up element bounds via pyatspi subprocess, finds the owning window
/// Looks up element bounds via native AT-SPI, finds the owning window
/// (via `xid_hint` or the first window for `pid`), then converts screen-absolute
/// → window-local coords via X11 translate_coordinates.
fn resolve_element_local_coords(pid: u32, idx: usize, xid_hint: Option<u64>)
Expand Down
Loading