From b48e2b6dd0341df945871dbf4eff8b474fe69514 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 17 Oct 2026 16:58:08 +0800 Subject: [PATCH 1/4] fix(computer-use): bound Windows UIA requests and preserve control status --- docs/architecture/computer-use-control.md | 20 +- scripts/test-windows-computer-use-native.mjs | 2 +- src/apps/desktop/README.md | 11 + src/apps/desktop/src/computer_use/AGENTS.md | 16 + src/apps/desktop/src/computer_use/mod.rs | 2 + .../src/computer_use/provider_worker.rs | 349 ++++++++++++ .../src/computer_use/windows_ax_shortcuts.rs | 27 +- .../desktop/src/computer_use/windows_ax_ui.rs | 496 +++++++++++------- .../ComputerUseControlCard.test.tsx | 36 ++ .../app/components/ComputerUseControlCard.tsx | 4 +- .../api/service-api/ComputerUseControlAPI.ts | 2 +- 11 files changed, 753 insertions(+), 212 deletions(-) create mode 100644 src/apps/desktop/src/computer_use/provider_worker.rs diff --git a/docs/architecture/computer-use-control.md b/docs/architecture/computer-use-control.md index 6708131c42..5f7a0e55a9 100644 --- a/docs/architecture/computer-use-control.md +++ b/docs/architecture/computer-use-control.md @@ -529,11 +529,29 @@ capture errors. No unlock, wake or automatic activation is a recovery action. ### Windows -Observation and semantic actions share a retained MTA UIA worker/cache. Validate +Observation, semantic actions, locate, hit-testing and shortcut discovery share +a retained MTA UIA worker/cache. Validate PID, HWND, generation, target revision, enabled state and geometry before supported Invoke, Toggle, SelectionItem or Scroll operations. Release retained COM references with the cache; do not rebuild indices during input. +Use `CUIAutomation8` / `IUIAutomation2` with automatic focus disabled, a two-second +connection timeout and a five-second transaction timeout. A separate ten-second +caller deadline bounds providers that do not honor COM timeouts. Control-token +cancellation is polled every 50 ms. An occupied worker rejects new requests with +`AX_PROVIDER_BUSY` until its native call actually returns; caller timeout never +creates replacement workers or queues more work. Stop and status do not wait on +that worker. Recheck the request fence immediately before input so a late +preflight cannot dispatch after abandonment. Once input enters a provider, a +timeout, cancellation or provider error means `INPUT_OUTCOME_UNKNOWN`; never +replay it or fall back to a coordinate click. Discard abandoned observations on +the owning apartment. Native input already submitted can still finish after +Stop, and a permanently hung provider requires recovery or a host restart. + +The control card treats preview availability separately from control status. +Missing AX-only images, capture failures and stale frames clear the preview +without hiding valid status or disabling Stop. + Standard Unicode Edit/RichEdit text insertion replaces the current selection through `EM_REPLACESEL`, preserving text outside that selection. An observed NodeIdx retains UIA identity. ImageXy resolves through the screenshot map and diff --git a/scripts/test-windows-computer-use-native.mjs b/scripts/test-windows-computer-use-native.mjs index b770fdaf48..e9d91f52db 100644 --- a/scripts/test-windows-computer-use-native.mjs +++ b/scripts/test-windows-computer-use-native.mjs @@ -71,7 +71,7 @@ pub mod util { pub mod errors { pub use crate::computer_use_contract::{ComputerUseContractError as OpenBitFunError, ComputerUseContractResult as OpenBitFunResult}; }} mod computer_use { - ${['control_session', 'process_identity', 'grants', 'permissions', 'windows_list_apps', 'windows_ax_ui', 'windows_msaa', 'windows_capture', 'windows_wgc_capture', 'windows_pointer_feedback', 'ui_locate_common', 'ax_snapshot_digest'].map(name => `#[path = ${source(`src/apps/desktop/src/computer_use/${name}.rs`)}] mod ${name};`).join('\n ')} + ${['control_session', 'process_identity', 'grants', 'permissions', 'windows_list_apps', 'windows_ax_ui', 'windows_ax_shortcuts', 'provider_worker', 'windows_msaa', 'windows_capture', 'windows_wgc_capture', 'windows_pointer_feedback', 'ui_locate_common', 'ax_snapshot_digest'].map(name => `#[path = ${source(`src/apps/desktop/src/computer_use/${name}.rs`)}] mod ${name};`).join('\n ')} mod desktop_host { use crate::util::errors::{OpenBitFunError, OpenBitFunResult}; #[path = ${source('src/apps/desktop/src/computer_use/desktop_host/background_guard.rs')}] mod background_guard; diff --git a/src/apps/desktop/README.md b/src/apps/desktop/README.md index 4d64fc1059..37ff281db2 100644 --- a/src/apps/desktop/README.md +++ b/src/apps/desktop/README.md @@ -218,6 +218,17 @@ Unsupported providers return a capability error. These Windows paths have cross-compilation coverage; interactive Windows acceptance is still pending. Windows does not have the macOS Sharing menu. +Windows accessibility requests return after ten seconds if the application's +provider stops responding. Stop remains available and prevents input that has +not yet been submitted. An `INPUT_OUTCOME_UNKNOWN` error means the application +may still complete the input: wait for it to recover, then observe before taking +another action. OpenBitFun does not automatically repeat the click or text. While +that native call is still running, new accessibility requests return +`AX_PROVIDER_BUSY`; a provider that never returns requires the target to recover +or OpenBitFun to restart. A missing preview only means no valid image is available +(including observations requested without images); it does not mean that the +control status is unavailable. + See the [cross-platform control contract](../../../docs/architecture/computer-use-control.md) for platform limits, lifecycle guarantees, and native verification scenarios. diff --git a/src/apps/desktop/src/computer_use/AGENTS.md b/src/apps/desktop/src/computer_use/AGENTS.md index 286345b825..0f8a9200ed 100644 --- a/src/apps/desktop/src/computer_use/AGENTS.md +++ b/src/apps/desktop/src/computer_use/AGENTS.md @@ -208,6 +208,22 @@ cargo build -p openbitfun-desktop ``` Windows-only paths (`windows_wgc_capture`, UIA) compile on CI (`windows-latest`). +UIA observation, semantic input, locate, hit-test and shortcut discovery share +one MTA worker. `provider_worker.rs` bounds admission and caller waits: a request +has a ten-second deadline and polls control cancellation every 50 ms. UIA also +uses two-second connection and five-second transaction limits with automatic +focus disabled. Never release the occupied slot, spawn replacement workers, or +queue another input just because a caller timed out. Foreign code may still be +running. Input that has crossed its submission fence has an unknown outcome on +failure; do not retry it or switch to a coordinate click. Stop fences input still +in preflight, while already submitted input can complete later. After the native +call returns, discard abandoned observations and require a fresh read. A provider +that never returns leaves UIA explicitly busy until it recovers or the host is +restarted; status and Stop must remain independent of that worker. + +The focused desktop `computer_use` tests include portable blocking-provider +regressions for deadlines, cancellation, late input, busy admission and recovery. +They simulate foreign-call stalls and do not establish real Windows delivery. The isolated native-module harness compiles the production control owner, UIA, MSAA, WGC and pointer modules. Its portable error-envelope adapter replaces no native APIs or algorithms. Cross-compilation is type-checking, not Windows diff --git a/src/apps/desktop/src/computer_use/mod.rs b/src/apps/desktop/src/computer_use/mod.rs index d8bd306973..ed707d9b3b 100644 --- a/src/apps/desktop/src/computer_use/mod.rs +++ b/src/apps/desktop/src/computer_use/mod.rs @@ -34,6 +34,8 @@ mod observation_cache; mod ocr_context; pub(crate) mod permissions; pub(crate) mod process_identity; +#[cfg(any(target_os = "windows", test))] +mod provider_worker; mod screen_ocr; pub(crate) mod share_grants; mod som_overlay; diff --git a/src/apps/desktop/src/computer_use/provider_worker.rs b/src/apps/desktop/src/computer_use/provider_worker.rs new file mode 100644 index 0000000000..bd0b15bcea --- /dev/null +++ b/src/apps/desktop/src/computer_use/provider_worker.rs @@ -0,0 +1,349 @@ +//! Bounded admission and cancellation for a thread-affine native provider. +//! +//! A caller deadline cannot interrupt foreign code. Keep the worker occupied +//! until that code actually returns, and fence input that has not been submitted. + +use std::sync::{ + atomic::{AtomicBool, AtomicU8, Ordering}, + mpsc::{self, RecvTimeoutError, SyncSender}, + Arc, Mutex, +}; +use std::time::{Duration, Instant}; + +const READING: u8 = 0; +const INPUT_SUBMITTED: u8 = 1; +const ABANDONED: u8 = 2; +const INPUT_UNKNOWN: u8 = 3; +const CANCEL_POLL: Duration = Duration::from_millis(50); + +pub(super) struct Request { + operation: &'static str, + started: Instant, + deadline: Instant, + state: AtomicU8, + phase: Mutex<&'static str>, +} + +impl Request { + fn new(operation: &'static str, timeout: Duration) -> Self { + let started = Instant::now(); + Self { + operation, + started, + deadline: started + timeout, + state: AtomicU8::new(READING), + phase: Mutex::new("queued"), + } + } + + pub(super) fn checkpoint(&self, phase: &'static str) -> Result<(), String> { + self.check()?; + *self.phase.lock().unwrap_or_else(|e| e.into_inner()) = phase; + Ok(()) + } + + pub(super) fn check(&self) -> Result<(), String> { + if Instant::now() >= self.deadline { + return Err(self.abandon("AX_PROVIDER_TIMEOUT")); + } + if self.state.load(Ordering::Acquire) >= ABANDONED { + return Err(self.abandon("AX_REQUEST_CANCELLED")); + } + Ok(()) + } + + /// Linearization point immediately before the single native input call. + /// Abandonment wins against input that is still in provider preflight. + pub(super) fn begin_input(&self, phase: &'static str) -> Result<(), String> { + self.checkpoint(phase)?; + self.state + .compare_exchange( + READING, + INPUT_SUBMITTED, + Ordering::AcqRel, + Ordering::Acquire, + ) + .map(|_| ()) + .map_err(|_| self.abandon("AX_REQUEST_CANCELLED")) + } + + /// A provider may report a transport failure after accepting the input. + /// Such an error is not evidence that retrying the action would be safe. + pub(super) fn input_failure(&self) -> Option { + matches!( + self.state.load(Ordering::Acquire), + INPUT_SUBMITTED | INPUT_UNKNOWN + ) + .then(|| self.failure_message("AX_NATIVE_INPUT_FAILED", true)) + } + + fn abandon(&self, reason: &str) -> String { + let previous = self + .state + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |state| { + Some(if state == INPUT_SUBMITTED || state == INPUT_UNKNOWN { + INPUT_UNKNOWN + } else { + ABANDONED + }) + }) + .unwrap(); + self.failure_message( + reason, + previous == INPUT_SUBMITTED || previous == INPUT_UNKNOWN, + ) + } + + fn failure_message(&self, reason: &str, input_unknown: bool) -> String { + let phase = *self.phase.lock().unwrap_or_else(|e| e.into_inner()); + let detail = format!( + "{reason}; operation={}, phase={}, elapsed_ms={}", + self.operation, + phase, + self.started.elapsed().as_millis() + ); + if input_unknown { + format!("[INPUT_OUTCOME_UNKNOWN] {detail}. Native input may still complete. Do not repeat the input; observe the target after the provider recovers. No retry was sent") + } else { + let code = if reason.starts_with("AX_") { + reason + } else { + "AX_REQUEST_CANCELLED" + }; + format!("[{code}] {detail}. No input was submitted. The provider must return before another UIA request can run") + } + } +} + +struct BusyPermit(Arc); +impl Drop for BusyPermit { + fn drop(&mut self) { + self.0.store(false, Ordering::Release); + } +} + +type Job = Box; + +pub(super) struct Worker { + sender: SyncSender>, + busy: Arc, +} + +impl Worker { + /// State is created, used and dropped on the provider thread; it need not + /// be Send. In particular, retained COM interfaces never cross apartments. + pub(super) fn new(initialize: impl FnOnce() -> S + Send + 'static) -> Self { + let (sender, receiver) = mpsc::sync_channel::>(1); + std::thread::spawn(move || { + let mut state = initialize(); + while let Ok(job) = receiver.recv() { + job(&mut state); + } + }); + Self { + sender, + busy: Arc::new(AtomicBool::new(false)), + } + } + + pub(super) fn run( + &self, + operation: &'static str, + timeout: Duration, + mut allowed: impl FnMut() -> Result<(), String>, + work: impl FnOnce(&mut S, Arc) -> Result + Send + 'static, + ) -> Result, String> { + allowed()?; + self.busy.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .map_err(|_| "[AX_PROVIDER_BUSY] The previous native UIA call has not returned. No request was queued and no input was sent; wait for the target application to recover".to_string())?; + let permit = BusyPermit(self.busy.clone()); + let request = Arc::new(Request::new(operation, timeout)); + let worker_request = request.clone(); + let (sender, receiver) = mpsc::sync_channel(1); + self.sender + .try_send(Box::new(move |state| { + let result = work(state, worker_request); + // Releasing admission here, rather than on caller timeout, bounds + // native threads and ensures that no second call overlaps this one. + drop(permit); + let _ = sender.send(result); + })) + .map_err(|_| "[AX_WORKER_UNAVAILABLE] UIA worker stopped".to_string())?; + loop { + if let Err(reason) = allowed() { + return Err(request.abandon(&reason)); + } + request.check()?; + let remaining = request.deadline.saturating_duration_since(Instant::now()); + match receiver.recv_timeout(remaining.min(CANCEL_POLL)) { + Ok(result) => { + if let Err(reason) = allowed() { + return Err(request.abandon(&reason)); + } + request.check()?; + return Ok(result); + } + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => { + return Err(request.abandon("AX_WORKER_UNAVAILABLE")) + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + + const TEST_TIMEOUT: Duration = Duration::from_millis(100); + + #[test] + fn blocked_preflight_times_out_without_late_input_or_queued_work() { + let worker = Worker::new(|| 0usize); + let (release, blocked) = mpsc::channel(); + let (done, completed) = mpsc::channel(); + let inputs = Arc::new(AtomicUsize::new(0)); + let input_count = inputs.clone(); + let result = worker.run( + "invoke", + TEST_TIMEOUT, + || Ok(()), + move |state, request| { + request.checkpoint("pattern_lookup")?; + blocked.recv().unwrap(); + let result = request.begin_input("invoke"); + if result.is_ok() { + input_count.fetch_add(1, Ordering::SeqCst); + } + *state += 1; + done.send(()).unwrap(); + result + }, + ); + assert!(result.unwrap_err().contains("AX_PROVIDER_TIMEOUT")); + assert_eq!(inputs.load(Ordering::SeqCst), 0); + let busy = worker.run( + "observe", + TEST_TIMEOUT, + || Ok(()), + |_, _| Ok::<_, String>(()), + ); + assert!(busy.unwrap_err().contains("AX_PROVIDER_BUSY")); + release.send(()).unwrap(); + completed.recv_timeout(Duration::from_secs(2)).unwrap(); + // Only the real provider completion releases admission. No replacement + // worker or queued action is allowed while the caller is timed out. + let start = Instant::now(); + while worker.busy.load(Ordering::Acquire) { + assert!(start.elapsed() < Duration::from_secs(2)); + std::thread::yield_now(); + } + assert_eq!( + worker + .run( + "observe", + TEST_TIMEOUT, + || Ok(()), + |state, _| Ok::<_, String>(*state) + ) + .unwrap() + .unwrap(), + 1 + ); + assert_eq!(inputs.load(Ordering::SeqCst), 0); + } + + #[test] + fn input_that_has_entered_the_provider_has_unknown_outcome_and_is_never_replayed() { + let worker = Worker::new(|| ()); + let (release, blocked) = mpsc::channel(); + let inputs = Arc::new(AtomicUsize::new(0)); + let input_count = inputs.clone(); + let result = worker.run( + "invoke", + TEST_TIMEOUT, + || Ok(()), + move |_, request| { + request.begin_input("invoke")?; + input_count.fetch_add(1, Ordering::SeqCst); + blocked.recv().unwrap(); + Ok::<_, String>(()) + }, + ); + assert!(result.unwrap_err().contains("INPUT_OUTCOME_UNKNOWN")); + assert_eq!(inputs.load(Ordering::SeqCst), 1); + release.send(()).unwrap(); + } + + #[test] + fn stop_releases_waiter_before_deadline_and_fences_late_preflight() { + let worker = Arc::new(Worker::new(|| ())); + let cancelled = Arc::new(AtomicBool::new(false)); + let caller_cancelled = cancelled.clone(); + let (entered, waiting) = mpsc::channel(); + let (release, blocked) = mpsc::channel(); + let (done, completed) = mpsc::channel(); + let caller = std::thread::spawn(move || { + worker.run( + "invoke", + Duration::from_secs(30), + || { + if caller_cancelled.load(Ordering::Acquire) { + Err("CONTROL_STOPPED".into()) + } else { + Ok(()) + } + }, + move |_, request| { + entered.send(()).unwrap(); + blocked.recv().unwrap(); + let result = request.begin_input("invoke"); + done.send(result.is_err()).unwrap(); + result + }, + ) + }); + waiting.recv_timeout(Duration::from_secs(2)).unwrap(); + let stopped = Instant::now(); + cancelled.store(true, Ordering::Release); + assert!(caller + .join() + .unwrap() + .unwrap_err() + .contains("CONTROL_STOPPED")); + assert!(stopped.elapsed() < Duration::from_secs(2)); + release.send(()).unwrap(); + assert!(completed.recv_timeout(Duration::from_secs(2)).unwrap()); + } + + #[test] + fn native_state_is_owned_by_one_thread_and_does_not_need_send() { + let worker = Worker::new(|| (std::rc::Rc::new(7), std::thread::current().id())); + let result = worker + .run( + "observe", + Duration::from_secs(2), + || Ok(()), + |state, request| { + request.check()?; + assert_eq!(state.1, std::thread::current().id()); + Ok::<_, String>(*state.0) + }, + ) + .unwrap() + .unwrap(); + assert_eq!(result, 7); + } + + #[test] + fn a_provider_error_after_input_does_not_claim_that_nothing_happened() { + let request = Request::new("invoke", Duration::from_secs(2)); + assert!(request.input_failure().is_none()); + request.begin_input("invoke").unwrap(); + let error = request.input_failure().unwrap(); + assert!(error.contains("INPUT_OUTCOME_UNKNOWN")); + assert!(request.begin_input("retry").is_err()); + } +} diff --git a/src/apps/desktop/src/computer_use/windows_ax_shortcuts.rs b/src/apps/desktop/src/computer_use/windows_ax_shortcuts.rs index b97d266a64..b78c6f1da2 100644 --- a/src/apps/desktop/src/computer_use/windows_ax_shortcuts.rs +++ b/src/apps/desktop/src/computer_use/windows_ax_shortcuts.rs @@ -22,14 +22,11 @@ use openbitfun_core::agentic::tools::computer_use_host::{ }; use openbitfun_core::util::errors::{OpenBitFunError, OpenBitFunResult}; use windows::Win32::Foundation::HWND; -use windows::Win32::System::Com::{ - CoCreateInstance, CoInitializeEx, CLSCTX_INPROC_SERVER, COINIT_APARTMENTTHREADED, -}; use windows::Win32::UI::Accessibility::{ - CUIAutomation, IUIAutomation, IUIAutomationCacheRequest, IUIAutomationElement, - IUIAutomationTogglePattern, ToggleState_On, TreeScope_Subtree, UIA_AcceleratorKeyPropertyId, - UIA_ControlTypePropertyId, UIA_IsEnabledPropertyId, UIA_MenuBarControlTypeId, - UIA_MenuItemControlTypeId, UIA_NamePropertyId, UIA_TogglePatternId, + IUIAutomation, IUIAutomationCacheRequest, IUIAutomationElement, IUIAutomationTogglePattern, + ToggleState_On, TreeScope_Subtree, UIA_AcceleratorKeyPropertyId, UIA_ControlTypePropertyId, + UIA_IsEnabledPropertyId, UIA_MenuBarControlTypeId, UIA_MenuItemControlTypeId, + UIA_NamePropertyId, UIA_TogglePatternId, }; /// Menu-tree recursion depth cap (menu bar → menu → item → submenu → …). @@ -221,16 +218,10 @@ fn find_menu_bar(root: &IUIAutomationElement) -> Option { /// found — modern Fluent/Electron apps frequently have no classic menu /// bar, which is a legitimate "no shortcuts (via this mechanism)" answer. pub(super) fn get_app_menu_shortcuts(hwnd: HWND) -> OpenBitFunResult<(Vec, u32)> { - unsafe { - let _ = CoInitializeEx(None, COINIT_APARTMENTTHREADED); - - let automation: IUIAutomation = - CoCreateInstance(&CUIAutomation, None, CLSCTX_INPROC_SERVER).map_err(|e| { - OpenBitFunError::tool(format!( - "UI Automation (CoCreateInstance CUIAutomation): {}.", - e - )) - })?; + let hwnd = hwnd.0 as isize; + super::windows_ax_ui::run_read("shortcuts", move || unsafe { + let hwnd = HWND(hwnd as *mut _); + let automation = super::windows_ax_ui::create_automation()?; let cache_req = build_shortcuts_cache_request(&automation)?; @@ -251,7 +242,7 @@ pub(super) fn get_app_menu_shortcuts(hwnd: HWND) -> OpenBitFunResult<(Vec return Ok(e), Err(e) => { attempt += 1; - if attempt >= BUILD_CACHE_MAX_ATTEMPTS { + if attempt >= BUILD_CACHE_MAX_ATTEMPTS + || e.code() != windows::Win32::Foundation::E_FAIL + { return Err(OpenBitFunError::tool(format!( "UI Automation BuildUpdatedCache failed after {} attempts: {}.", attempt, e @@ -432,22 +431,13 @@ unsafe fn walk_tree_full( max_elements: usize, max_depth: usize, ) -> OpenBitFunResult<(String, Vec)> { - // SAFETY: initializes COM for the current thread and creates the documented - // in-process UI Automation class; failures are handled below. - let _ = unsafe { CoInitializeEx(None, COINIT_APARTMENTTHREADED) }; - - let automation: IUIAutomation = - unsafe { CoCreateInstance(&CUIAutomation, None, CLSCTX_INPROC_SERVER) }.map_err(|e| { - OpenBitFunError::tool(format!( - "UI Automation (CoCreateInstance CUIAutomation): {}.", - e - )) - })?; + let automation = create_automation()?; let cache_req = unsafe { build_cache_request(&automation) }?; // SAFETY: `hwnd` is the caller-provided target handle. UIA reports an // error for an invalid or stale handle, which is propagated here. + checkpoint("element_from_handle")?; let uncached = unsafe { automation.ElementFromHandle(hwnd) }.map_err(|e| { OpenBitFunError::tool(format!("UI Automation ElementFromHandle failed: {}.", e)) })?; @@ -488,7 +478,10 @@ unsafe fn walk_cached_bounded( max_elements: usize, max_depth: usize, ) { - if depth > max_depth || *total >= max_elements { + if depth > max_depth + || *total >= max_elements + || request_checkpoint("walk_cached_tree").is_err() + { return; } *total += 1; @@ -770,6 +763,17 @@ pub(super) fn locate_ui_element_center_for_window( hwnd_raw: isize, expected_pid: u32, query: &UiElementLocateQuery, +) -> OpenBitFunResult { + let query = query.clone(); + run_read("locate", move || { + locate_ui_element_center_for_window_native(hwnd_raw, expected_pid, &query) + }) +} + +fn locate_ui_element_center_for_window_native( + hwnd_raw: isize, + expected_pid: u32, + query: &UiElementLocateQuery, ) -> OpenBitFunResult { ui_locate_common::validate_query(query)?; @@ -869,6 +873,17 @@ pub(super) fn accessibility_hit_at_global_point_for_window( expected_pid: u32, gx: f64, gy: f64, +) -> OpenBitFunResult> { + run_read("hit_test", move || { + accessibility_hit_at_global_point_for_window_native(hwnd_raw, expected_pid, gx, gy) + }) +} + +fn accessibility_hit_at_global_point_for_window_native( + hwnd_raw: isize, + expected_pid: u32, + gx: f64, + gy: f64, ) -> OpenBitFunResult> { let hwnd = HWND(hwnd_raw as *mut _); if !gx.is_finite() || !gy.is_finite() || !target_window_matches(hwnd, expected_pid) { @@ -896,9 +911,8 @@ pub(super) fn accessibility_hit_at_global_point_for_window( if unsafe { element.CurrentProcessId() }.ok() != Some(expected_pid as i32) { return Ok(None); } - let automation: IUIAutomation = - unsafe { CoCreateInstance(&CUIAutomation, None, CLSCTX_INPROC_SERVER) } - .map_err(|error| OpenBitFunError::tool(format!("UI Automation: {error}")))?; + let automation = create_automation()?; + checkpoint("hit_test_root")?; let root = unsafe { automation.ElementFromHandle(hwnd) } .map_err(|error| OpenBitFunError::tool(format!("UI Automation target root: {error}")))?; let walker = unsafe { automation.RawViewWalker() } @@ -906,6 +920,7 @@ pub(super) fn accessibility_hit_at_global_point_for_window( let mut current = element; let mut belongs_to_window = false; for _ in 0..200 { + checkpoint("hit_test_ancestry")?; if unsafe { automation.CompareElements(¤t, &root) }.is_ok_and(|same| same.as_bool()) { belongs_to_window = true; break; @@ -970,6 +985,7 @@ fn snapshot_and_nodes( { Ok((text, nodes)) if !nodes.is_empty() => (text, nodes), primary => { + checkpoint("legacy_observation_fallback")?; if crate::computer_use::windows_msaa::is_sal_vcl_window(hwnd_raw) { match crate::computer_use::windows_msaa::walk_msaa_tree(hwnd_raw) { Ok(msaa_nodes) if !msaa_nodes.is_empty() => { @@ -1102,129 +1118,219 @@ enum SemanticAction { Scroll(u32, i32, i32), Center(u32), } -enum UiaRequest { - Clear, - InsertNative( - isize, - Option<(f64, f64)>, - String, - super::control_session::ControlToken, - std::sync::mpsc::SyncSender>, - ), - Observe( - isize, - u32, - bool, - super::control_session::ControlToken, - std::sync::mpsc::SyncSender>, - ), - Act( - isize, - SemanticAction, - super::control_session::ControlToken, - std::sync::mpsc::SyncSender>, - ), -} struct ObservedWindow { hwnd: isize, scope: openbitfun_agent_tools::computer_use_control::ObservationScope, pid: u32, nodes: Vec, } -static UIA_WORKER: std::sync::OnceLock> = +struct WorkerState { + initialized: bool, + observed: Option, +} +impl Drop for WorkerState { + fn drop(&mut self) { + self.observed = None; + if self.initialized { + unsafe { windows::Win32::System::Com::CoUninitialize() }; + } + } +} +// Two seconds for connection, five for one provider transaction and ten for +// the complete request (including retries). COM limits alone do not bound a +// misbehaving Invoke provider, so the caller also enforces the outer deadline. +const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +static UIA_WORKER: std::sync::OnceLock> = std::sync::OnceLock::new(); +static CLEAR_OBSERVATIONS: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); +thread_local! { + static CURRENT_REQUEST: std::cell::RefCell>> = const { std::cell::RefCell::new(None) }; +} pub(super) fn clear_observations() { - if let Some(worker) = UIA_WORKER.get() { - let _ = worker.send(UiaRequest::Clear); + // Never enqueue behind a stuck provider, and never release COM on the + // cancelling thread. The worker discards it before accepting further work. + CLEAR_OBSERVATIONS.store(true, std::sync::atomic::Ordering::Release); +} + +pub(super) fn checkpoint(phase: &'static str) -> OpenBitFunResult<()> { + super::control_session::capture_allowed().map_err(OpenBitFunError::tool)?; + request_checkpoint(phase) +} + +// Cached traversal checks the caller's fence without repeatedly opening the +// authorized process. Full authority is rechecked before publication and input. +fn request_checkpoint(phase: &'static str) -> OpenBitFunResult<()> { + CURRENT_REQUEST.with_borrow(|request| { + request + .as_ref() + .ok_or_else(|| { + OpenBitFunError::tool("[AX_WORKER_REQUIRED] UIA requires the provider worker") + })? + .checkpoint(phase) + .map_err(OpenBitFunError::tool) + }) +} +fn begin_input(phase: &'static str) -> OpenBitFunResult<()> { + super::control_session::input_allowed().map_err(OpenBitFunError::tool)?; + CURRENT_REQUEST.with_borrow(|request| { + request + .as_ref() + .ok_or_else(|| { + OpenBitFunError::tool("[AX_WORKER_REQUIRED] UIA input requires the provider worker") + })? + .begin_input(phase) + .map_err(OpenBitFunError::tool) + }) +} + +fn current_pattern( + element: &IUIAutomationElement, + pattern_id: windows::Win32::UI::Accessibility::UIA_PATTERN_ID, + phase: &'static str, +) -> OpenBitFunResult> { + checkpoint(phase)?; + let pattern = unsafe { element.GetCurrentPatternAs::(pattern_id) }; + // A failed lookup can return long after cancellation. Do not proceed to + // another pattern lookup (or another native input route) in that case. + checkpoint(phase)?; + Ok(pattern.ok()) +} + +pub(super) fn create_automation() -> OpenBitFunResult { + checkpoint("create_automation")?; + let configured: IUIAutomation2 = + unsafe { CoCreateInstance(&CUIAutomation8, None, CLSCTX_INPROC_SERVER) }.map_err( + |error| OpenBitFunError::tool(format!("UI Automation initialization: {error}")), + )?; + unsafe { + configured + .SetAutoSetFocus(false) + .and_then(|_| configured.SetConnectionTimeout(2000)) + .and_then(|_| configured.SetTransactionTimeout(5000)) } + .map_err(|error| { + OpenBitFunError::tool(format!( + "UI Automation timeout/focus configuration: {error}" + )) + })?; + configured + .cast() + .map_err(|error| OpenBitFunError::tool(format!("UI Automation interface: {error}"))) } -fn uia_worker() -> &'static std::sync::mpsc::Sender { - UIA_WORKER.get_or_init(|| { - let (tx, rx) = std::sync::mpsc::channel(); - std::thread::spawn(move || { - // UIA client work belongs on an MTA thread separate from the UI. - let initialized = unsafe { CoInitializeEx(None, windows::Win32::System::Com::COINIT_MULTITHREADED) }.is_ok(); - let mut observed: Option = None; - while let Ok(request) = rx.recv() { - match request { - UiaRequest::Clear => { observed = None; }, - UiaRequest::InsertNative(hwnd, point, text, token, reply) => { - let result = super::control_session::with_token(token, || insert_native_target(hwnd, point, &text)); - let _ = reply.send(result); +fn run_uia( + operation: &'static str, + work: impl FnOnce(&mut WorkerState) -> OpenBitFunResult + Send + 'static, +) -> OpenBitFunResult { + let token = super::control_session::capture_token().map_err(OpenBitFunError::tool)?; + let worker = UIA_WORKER.get_or_init(|| { + super::provider_worker::Worker::new(|| WorkerState { + initialized: unsafe { CoInitializeEx(None, COINIT_MULTITHREADED) }.is_ok(), + observed: None, + }) + }); + let result = worker + .run( + operation, + REQUEST_TIMEOUT, + || super::control_session::with_token(token, super::control_session::capture_allowed), + move |state, request| { + super::control_session::with_token(token, || { + struct Reset; + impl Drop for Reset { + fn drop(&mut self) { + CURRENT_REQUEST.with_borrow_mut(|r| *r = None); + } } - UiaRequest::Observe(hwnd, depth, focus, token, reply) => { - let result = super::control_session::with_token(token, || { - super::control_session::capture_allowed().map_err(OpenBitFunError::tool)?; - let scope = super::control_session::observation_scope().map_err(OpenBitFunError::tool)?; - let handle = windows::Win32::Foundation::HWND(hwnd as *mut _); - let (snapshot, nodes) = snapshot_and_nodes(handle, depth, focus)?; - super::control_session::capture_allowed().map_err(OpenBitFunError::tool)?; - super::control_session::validate_observation_scope(&scope).map_err(OpenBitFunError::tool)?; - observed = Some(ObservedWindow { hwnd, scope, - pid: window_pid_for(handle).unwrap_or(0), nodes }); - Ok(snapshot) - }); - if result.is_err() { observed = None; } - let _ = reply.send(result); + CURRENT_REQUEST.with_borrow_mut(|r| *r = Some(request.clone())); + let _reset = Reset; + if CLEAR_OBSERVATIONS.swap(false, std::sync::atomic::Ordering::AcqRel) { + state.observed = None; } - UiaRequest::Act(hwnd, action, token, reply) => { - let result = super::control_session::with_token(token, || { - let cache = observed.as_ref().ok_or_else(|| OpenBitFunError::tool( - "[AX_OBSERVATION_REQUIRED] Read the target window before using a node index"))?; - apply_semantic(cache, hwnd, action) - }); - let _ = reply.send(result); + let result = (|| { + if !state.initialized { + return Err(OpenBitFunError::tool( + "UI Automation MTA initialization failed", + )); + } + checkpoint("started")?; + log::debug!("UIA request started: operation={operation}"); + let output = work(state)?; + checkpoint("completed")?; + Ok(output) + })() + .map_err(|error| { + if let Some(reason) = request.input_failure() { + log::warn!("UIA input failed after submission: {reason}"); + OpenBitFunError::tool(format!("{reason}; provider error: {error}")) + } else { + error + } + }); + if (result.is_err() && request.input_failure().is_some()) + || request.check().is_err() + || super::control_session::capture_allowed().is_err() + || CLEAR_OBSERVATIONS.swap(false, std::sync::atomic::Ordering::AcqRel) + { + state.observed = None; } - } - } - drop(observed); - if initialized { unsafe { windows::Win32::System::Com::CoUninitialize() }; } - }); - tx - }) + result + }) + }, + ) + .map_err(|reason| { + log::warn!("UIA request did not complete: {reason}"); + OpenBitFunError::tool(reason) + })?; + if result.is_ok() { + log::debug!("UIA request completed: operation={operation}"); + } + result +} + +pub(super) fn run_read( + operation: &'static str, + work: impl FnOnce() -> OpenBitFunResult + Send + 'static, +) -> OpenBitFunResult { + run_uia(operation, move |_| work()) } + pub(super) fn get_app_state_snapshot_for_window( - hwnd: windows::Win32::Foundation::HWND, + hwnd: HWND, max_depth: u32, focus_window_only: bool, ) -> OpenBitFunResult { - let token = super::control_session::capture_token().map_err(OpenBitFunError::tool)?; - let (tx, rx) = std::sync::mpsc::sync_channel(1); - uia_worker() - .send(UiaRequest::Observe( - hwnd.0 as isize, - max_depth, - focus_window_only, - token, - tx, - )) - .map_err(|_| OpenBitFunError::tool("UIA worker unavailable"))?; - rx.recv() - .map_err(|_| OpenBitFunError::tool("UIA worker stopped"))? + let hwnd = hwnd.0 as isize; + run_uia("observe", move |state| { + state.observed = None; + let scope = super::control_session::observation_scope().map_err(OpenBitFunError::tool)?; + let handle = HWND(hwnd as *mut _); + let (snapshot, nodes) = snapshot_and_nodes(handle, max_depth, focus_window_only)?; + checkpoint("publish_observation")?; + super::control_session::validate_observation_scope(&scope) + .map_err(OpenBitFunError::tool)?; + state.observed = Some(ObservedWindow { + hwnd, + scope, + pid: window_pid_for(handle).unwrap_or(0), + nodes, + }); + Ok(snapshot) + }) } /// Resolve an image-derived point inside the bound native window, or that -/// window thread's existing focus. This never asks the desktop which app is -/// under the real pointer, and never moves keyboard focus. +/// window thread's existing focus, without moving keyboard focus. pub(super) fn insert_text_at_bound_target( hwnd: isize, point: Option<(f64, f64)>, text: &str, ) -> OpenBitFunResult<()> { - let token = super::control_session::capture_token().map_err(OpenBitFunError::tool)?; - let (tx, rx) = std::sync::mpsc::sync_channel(1); - uia_worker() - .send(UiaRequest::InsertNative( - hwnd, - point, - text.into(), - token, - tx, - )) - .map_err(|_| OpenBitFunError::tool("UIA worker unavailable"))?; - rx.recv() - .map_err(|_| OpenBitFunError::tool("UIA worker stopped"))? + let text = text.to_owned(); + run_read("insert_native", move || { + insert_native_target(hwnd, point, &text) + }) } fn point_in_window(point: (f64, f64), rect: (i32, i32, i32, i32)) -> bool { @@ -1303,17 +1409,19 @@ fn insert_native_target( } info.hwndFocus }; - let automation: IUIAutomation = - CoCreateInstance(&CUIAutomation, None, CLSCTX_INPROC_SERVER).map_err(native_error)?; + let automation = create_automation()?; + checkpoint("native_text_element")?; let element = automation.ElementFromHandle(child).map_err(native_error)?; // The retained UIA element must still identify the precise child that // was hit or focused; providers must not redirect us to another HWND. + checkpoint("native_text_identity")?; if element.CurrentNativeWindowHandle().map_err(native_error)? != child { return Err(OpenBitFunError::tool( "[AX_STALE_ELEMENT] UIA native identity differs from the resolved text control", )); } if let Some(point) = point { + checkpoint("native_text_geometry")?; let rect = element.CurrentBoundingRectangle().map_err(native_error)?; if !point_in_window(point, (rect.left, rect.top, rect.right, rect.bottom)) { return Err(OpenBitFunError::tool( @@ -1363,12 +1471,14 @@ fn replace_native_edit_selection( ES_READONLY, GA_ROOT, GWL_STYLE, SMTO_ABORTIFHUNG, SMTO_BLOCK, }; let buffer = edit_replacement_utf16(text).map_err(OpenBitFunError::tool)?; + checkpoint("native_edit_preflight")?; unsafe { let child = element.CurrentNativeWindowHandle().map_err(|_| { OpenBitFunError::tool( "[BACKGROUND_TEXT_UNAVAILABLE] Observed UIA element has no native edit window", ) })?; + checkpoint("native_edit_process")?; if element .CurrentProcessId() .map_err(|e| OpenBitFunError::tool(format!("[AX_ACTION_FAILED] {e}")))? @@ -1395,9 +1505,12 @@ fn replace_native_edit_selection( "[BACKGROUND_TEXT_UNAVAILABLE] Native edit control is disabled or read-only", )); } - if let Ok(pattern) = - element.GetCurrentPatternAs::(UIA_ValuePatternId) - { + if let Some(pattern) = current_pattern::( + element, + UIA_ValuePatternId, + "native_edit_value_pattern", + )? { + checkpoint("native_edit_read_only")?; if pattern .CurrentIsReadOnly() .map_err(|e| OpenBitFunError::tool(format!("[AX_ACTION_FAILED] {e}")))? @@ -1416,6 +1529,7 @@ fn replace_native_edit_selection( .map_err(OpenBitFunError::tool)?; // This system message is marshalled by Windows across processes. Its // result has no success value; only the transport completion is known. + begin_input("native_edit_replace_selection")?; let completed = SendMessageTimeoutW( child, 0x00C2, @@ -1438,13 +1552,20 @@ fn replace_native_edit_selection( } fn dispatch_semantic(hwnd: isize, action: SemanticAction) -> OpenBitFunResult<(f64, f64)> { - let token = super::control_session::capture_token().map_err(OpenBitFunError::tool)?; - let (tx, rx) = std::sync::mpsc::sync_channel(1); - uia_worker() - .send(UiaRequest::Act(hwnd, action, token, tx)) - .map_err(|_| OpenBitFunError::tool("UIA worker unavailable"))?; - rx.recv() - .map_err(|_| OpenBitFunError::tool("UIA worker stopped"))? + let operation = match &action { + SemanticAction::Invoke(_) => "invoke", + SemanticAction::Insert(_, _) => "insert", + SemanticAction::Scroll(_, _, _) => "scroll", + SemanticAction::Center(_) => "center", + }; + run_uia(operation, move |state| { + let cache = state.observed.as_ref().ok_or_else(|| { + OpenBitFunError::tool( + "[AX_OBSERVATION_REQUIRED] Read the target window before using a node index", + ) + })?; + apply_semantic(cache, hwnd, action) + }) } pub(super) fn invoke_cached_node(hwnd: isize, idx: u32) -> OpenBitFunResult<()> { dispatch_semantic(hwnd, SemanticAction::Invoke(idx)).map(|_| ()) @@ -1499,14 +1620,16 @@ fn apply_semantic( ) })?; let native = |e: windows::core::Error| OpenBitFunError::tool(format!("[AX_ACTION_FAILED] {e}")); + checkpoint("semantic_preflight")?; unsafe { - if element.CurrentProcessId().map_err(native)? as u32 != cache.pid - || !element.CurrentIsEnabled().map_err(native)?.as_bool() - { + let pid = element.CurrentProcessId().map_err(native)? as u32; + checkpoint("semantic_enabled")?; + if pid != cache.pid || !element.CurrentIsEnabled().map_err(native)?.as_bool() { return Err(OpenBitFunError::tool( "[AX_STALE_ELEMENT] The observed control is no longer available", )); } + checkpoint("semantic_geometry")?; let rect = element.CurrentBoundingRectangle().map_err(native)?; if node.rect != Some((rect.left, rect.top, rect.right, rect.bottom)) { return Err(OpenBitFunError::tool( @@ -1524,22 +1647,29 @@ fn apply_semantic( let clicked = matches!(action, SemanticAction::Invoke(_)); match action { SemanticAction::Invoke(_) => { - if let Ok(pattern) = - element.GetCurrentPatternAs::(UIA_InvokePatternId) - { + if let Some(pattern) = current_pattern::( + &element, + UIA_InvokePatternId, + "invoke_pattern", + )? { super::control_session::input_allowed().map_err(OpenBitFunError::tool)?; + begin_input("invoke")?; pattern.Invoke().map_err(native)?; - } else if let Ok(pattern) = - element.GetCurrentPatternAs::(UIA_TogglePatternId) - { + } else if let Some(pattern) = current_pattern::( + &element, + UIA_TogglePatternId, + "toggle_pattern", + )? { super::control_session::input_allowed().map_err(OpenBitFunError::tool)?; + begin_input("toggle")?; pattern.Toggle().map_err(native)?; - } else if let Ok(pattern) = element - .GetCurrentPatternAs::( - UIA_SelectionItemPatternId, - ) - { + } else if let Some(pattern) = current_pattern::( + &element, + UIA_SelectionItemPatternId, + "selection_pattern", + )? { super::control_session::input_allowed().map_err(OpenBitFunError::tool)?; + begin_input("select")?; pattern.Select().map_err(native)?; } else { return Err(OpenBitFunError::tool("[AX_ACTION_UNSUPPORTED] Control exposes no invoke, toggle or selection pattern")); @@ -1549,13 +1679,16 @@ fn apply_semantic( replace_native_edit_selection(&element, HWND(hwnd as *mut _), cache.pid, &text)?; } SemanticAction::Scroll(_, dx, dy) => { - let pattern = element - .GetCurrentPatternAs::(UIA_ScrollPatternId) - .map_err(|_| { - OpenBitFunError::tool( - "[AX_ACTION_UNSUPPORTED] Control has no background Scroll pattern", - ) - })?; + let pattern = current_pattern::( + &element, + UIA_ScrollPatternId, + "scroll_pattern", + )? + .ok_or_else(|| { + OpenBitFunError::tool( + "[AX_ACTION_UNSUPPORTED] Control has no background Scroll pattern", + ) + })?; let amount = |delta: i32| { if delta > 0 { ScrollAmount_SmallIncrement @@ -1566,10 +1699,12 @@ fn apply_semantic( } }; super::control_session::input_allowed().map_err(OpenBitFunError::tool)?; + begin_input("scroll")?; pattern.Scroll(amount(dx), amount(dy)).map_err(native)?; } SemanticAction::Center(_) => unreachable!(), } + checkpoint("record_pointer")?; super::control_session::record_pointer(center.0, center.1, clicked); Ok(center) } @@ -1810,54 +1945,35 @@ mod control_native_tests { // RichEdit may expose a Text provider without a Value pattern. The // same native edit message still preserves its exact selection. unsafe { - use windows::Win32::System::Com::{CoUninitialize, COINIT_MULTITHREADED}; - CoInitializeEx(None, COINIT_MULTITHREADED).ok().unwrap(); - let automation: IUIAutomation = - CoCreateInstance(&CUIAutomation, None, CLSCTX_INPROC_SERVER).unwrap(); - let rich_element = automation.ElementFromHandle(HWND(rich as *mut _)).unwrap(); - replace_native_edit_selection( - &rich_element, - HWND(hwnd as *mut _), - std::process::id(), - "prefix \u{4e2d}\u{6587}", - ) - .unwrap(); - replace_native_edit_selection( - &rich_element, - HWND(hwnd as *mut _), - std::process::id(), - "\u{1f642}", - ) - .unwrap(); + let replace_rich = |top: isize, text: &str| { + let text = text.to_owned(); + run_read("rich_edit_fixture", move || { + let automation = create_automation()?; + let element = automation.ElementFromHandle(HWND(rich as *mut _)).unwrap(); + replace_native_edit_selection( + &element, + HWND(top as *mut _), + std::process::id(), + &text, + ) + }) + }; + replace_rich(hwnd, "prefix \u{4e2d}\u{6587}").unwrap(); + replace_rich(hwnd, "\u{1f642}").unwrap(); SendMessageW( HWND(rich as *mut _), 0x00B1, Some(WPARAM(0)), Some(LPARAM(7)), ); - replace_native_edit_selection( - &rich_element, - HWND(hwnd as *mut _), - std::process::id(), - "selected ", - ) - .unwrap(); + replace_rich(hwnd, "selected ").unwrap(); let mut buffer = [0u16; 64]; let count = GetWindowTextW(HWND(rich as *mut _), &mut buffer); assert_eq!( String::from_utf16_lossy(&buffer[..count as usize]), "selected \u{4e2d}\u{6587}\u{1f642}" ); - assert!(replace_native_edit_selection( - &rich_element, - HWND(toggle as *mut _), - std::process::id(), - "wrong bound window" - ) - .is_err()); - drop(rich_element); - drop(automation); - CoUninitialize(); + assert!(replace_rich(toggle, "wrong bound window").is_err()); } let mut thread_state = GUITHREADINFO { cbSize: std::mem::size_of::() as u32, diff --git a/src/web-ui/src/app/components/ComputerUseControlCard.test.tsx b/src/web-ui/src/app/components/ComputerUseControlCard.test.tsx index 2ccd776179..955a7231f3 100644 --- a/src/web-ui/src/app/components/ComputerUseControlCard.test.tsx +++ b/src/web-ui/src/app/components/ComputerUseControlCard.test.tsx @@ -160,3 +160,39 @@ it('discards a preview from an earlier visit to the same target', async () => { expect(container.querySelector('img')).toBeNull(); expect(container.textContent).toContain('computerControl.previewUnavailable'); }); + +it.each(['absent', 'rejected', 'wrong-target', 'stale-revision', 'invalid-image'])( + 'keeps live control status and Stop available when the preview is %s', async kind => { + mocks.status.mockResolvedValue({ ...active('Window A'), target_revision: 2 }); + const frame = { generation: 4, target: 'Window A', target_revision: 2, + image_base64: 'AA==', mime_type: 'image/png', width: 100, height: 100 }; + if (kind === 'rejected') mocks.preview.mockRejectedValue(new Error('Capture unavailable')); + else mocks.preview.mockResolvedValue(kind === 'absent' ? null : { + ...frame, + ...(kind === 'wrong-target' ? { target: 'Window B' } : {}), + ...(kind === 'stale-revision' ? { target_revision: 1 } : {}), + ...(kind === 'invalid-image' ? { width: 0 } : {}), + }); + mocks.stop.mockResolvedValue({ ...active('Window A', 5), state: 'stopped' }); + await act(async () => root.render()); + await act(async () => [...container.querySelectorAll('button')].find(b => b.textContent === 'computerControl.showPreview')!.click()); + await act(async () => vi.advanceTimersByTime(1000)); + expect(container.querySelector('img')).toBeNull(); + expect(container.textContent).toContain('computerControl.previewUnavailable'); + expect(container.textContent).toContain('computerControl.background'); + expect(container.textContent).not.toContain('computerControl.unavailable'); + await act(async () => [...container.querySelectorAll('button')].find(b => b.textContent === 'computerControl.stop')!.click()); + expect(mocks.stop).toHaveBeenCalledWith(4); + expect(container.textContent).toContain('computerControl.stopped'); + }, +); + +it('still reports a failed status read as unavailable and permits Stop', async () => { + mocks.status.mockResolvedValueOnce(active('Window A')).mockRejectedValue(new Error('Host unavailable')); + mocks.stop.mockResolvedValue({ ...active('Window A', 5), state: 'stopped' }); + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTime(1000)); + expect(container.textContent).toContain('computerControl.unavailable'); + await act(async () => [...container.querySelectorAll('button')].find(b => b.textContent === 'computerControl.stop')!.click()); + expect(mocks.stop).toHaveBeenCalledWith(4); +}); diff --git a/src/web-ui/src/app/components/ComputerUseControlCard.tsx b/src/web-ui/src/app/components/ComputerUseControlCard.tsx index 3b5dd665c5..608f55e63e 100644 --- a/src/web-ui/src/app/components/ComputerUseControlCard.tsx +++ b/src/web-ui/src/app/components/ComputerUseControlCard.tsx @@ -72,8 +72,10 @@ export function ComputerUseControlCard() { && (previous.snapshot.target_revision ?? 0) === (next.target_revision ?? 0) ? { ...previous, frame, framePointer: next.pointer } : previous); } catch { + // Preview is optional (for example, an AX-only observation). A + // missing, stale or failed frame does not invalidate live status. if (!disposed && scope.isCurrent() && revision === mutation.current) setView(previous => previous && { - ...previous, frame: undefined, framePointer: undefined, error: true, + ...previous, frame: undefined, framePointer: undefined, }); } } diff --git a/src/web-ui/src/infrastructure/api/service-api/ComputerUseControlAPI.ts b/src/web-ui/src/infrastructure/api/service-api/ComputerUseControlAPI.ts index e758a7dd74..e894c84952 100644 --- a/src/web-ui/src/infrastructure/api/service-api/ComputerUseControlAPI.ts +++ b/src/web-ui/src/infrastructure/api/service-api/ComputerUseControlAPI.ts @@ -124,7 +124,7 @@ export const computerUseControlAPI = { stop: (generation: number) => api.invoke( 'computer_use_control_stop', { request: { generation } }, ), - preview: (generation: number) => api.invoke( + preview: (generation: number) => api.invoke( 'computer_use_control_preview', { request: { generation } }, ), permissions: () => api.invoke<{ permissions?: ComputerUsePermissions }>('computer_use_get_status'), From 189ffb350f332d00834a58314245d4ed99dd8a33 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 17 Oct 2026 17:04:04 +0800 Subject: [PATCH 2/4] fix(computer-use): reject observations whose replies were abandoned --- docs/architecture/computer-use-control.md | 4 +- .../src/computer_use/provider_worker.rs | 82 ++++++++++++++++++- .../desktop/src/computer_use/windows_ax_ui.rs | 14 ++++ 3 files changed, 98 insertions(+), 2 deletions(-) diff --git a/docs/architecture/computer-use-control.md b/docs/architecture/computer-use-control.md index 5f7a0e55a9..51059ce225 100644 --- a/docs/architecture/computer-use-control.md +++ b/docs/architecture/computer-use-control.md @@ -545,7 +545,9 @@ that worker. Recheck the request fence immediately before input so a late preflight cannot dispatch after abandonment. Once input enters a provider, a timeout, cancellation or provider error means `INPUT_OUTCOME_UNKNOWN`; never replay it or fall back to a coordinate click. Discard abandoned observations on -the owning apartment. Native input already submitted can still finish after +the owning apartment. A retained observation also carries caller acceptance: +publishing native cache entries before a rejected reply does not make their +indices valid for subsequent input. Native input already submitted can still finish after Stop, and a permanently hung provider requires recovery or a host restart. The control card treats preview availability separately from control status. diff --git a/src/apps/desktop/src/computer_use/provider_worker.rs b/src/apps/desktop/src/computer_use/provider_worker.rs index bd0b15bcea..7c9655c5b9 100644 --- a/src/apps/desktop/src/computer_use/provider_worker.rs +++ b/src/apps/desktop/src/computer_use/provider_worker.rs @@ -21,6 +21,7 @@ pub(super) struct Request { started: Instant, deadline: Instant, state: AtomicU8, + response_accepted: AtomicBool, phase: Mutex<&'static str>, } @@ -32,6 +33,7 @@ impl Request { started, deadline: started + timeout, state: AtomicU8::new(READING), + response_accepted: AtomicBool::new(false), phase: Mutex::new("queued"), } } @@ -42,6 +44,13 @@ impl Request { Ok(()) } + /// A native observation may be cached just before the caller rejects its + /// reply. Only the caller can publish acceptance; an accepted observation + /// remains usable beyond the completed request's execution deadline. + pub(super) fn response_accepted(&self) -> bool { + self.response_accepted.load(Ordering::Acquire) + } + pub(super) fn check(&self) -> Result<(), String> { if Instant::now() >= self.deadline { return Err(self.abandon("AX_PROVIDER_TIMEOUT")); @@ -181,6 +190,9 @@ impl Worker { return Err(request.abandon(&reason)); } request.check()?; + if result.is_ok() { + request.response_accepted.store(true, Ordering::Release); + } return Ok(result); } Err(RecvTimeoutError::Timeout) => {} @@ -197,7 +209,7 @@ mod tests { use super::*; use std::sync::atomic::AtomicUsize; - const TEST_TIMEOUT: Duration = Duration::from_millis(100); + const TEST_TIMEOUT: Duration = Duration::from_secs(1); #[test] fn blocked_preflight_times_out_without_late_input_or_queued_work() { @@ -346,4 +358,72 @@ mod tests { assert!(error.contains("INPUT_OUTCOME_UNKNOWN")); assert!(request.begin_input("retry").is_err()); } + + #[test] + fn an_observation_cached_before_a_rejected_reply_is_never_accepted() { + let worker = Worker::new(|| None::>); + let cancelled = Arc::new(AtomicBool::new(false)); + let provider_cancelled = cancelled.clone(); + let (published, cached) = mpsc::sync_channel(1); + let result = worker.run( + "observe", + Duration::from_secs(5), + || { + if cancelled.load(Ordering::Acquire) { + Err("CONTROL_STOPPED".into()) + } else { + Ok(()) + } + }, + move |state, request| { + *state = Some(request.clone()); + published.send(request).unwrap(); + // Publication on the native thread does not guarantee that + // the reply survived the caller's final authority check. + provider_cancelled.store(true, Ordering::Release); + Ok::<_, String>(()) + }, + ); + assert!(result.unwrap_err().contains("CONTROL_STOPPED")); + assert!(!cached + .recv_timeout(Duration::from_secs(2)) + .unwrap() + .response_accepted()); + } + + #[test] + fn only_a_successful_reply_is_accepted_for_retained_observations() { + let worker = Worker::new(|| ()); + let request = worker + .run( + "observe", + Duration::from_secs(5), + || Ok(()), + |_, request| { + assert!(!request.response_accepted()); + Ok::<_, String>(request) + }, + ) + .unwrap() + .unwrap(); + assert!(request.response_accepted()); + let (published, cached) = mpsc::sync_channel(1); + let error = worker + .run( + "observe", + Duration::from_secs(5), + || Ok(()), + move |_, request| { + published.send(request).unwrap(); + Err::<(), _>("provider failure") + }, + ) + .unwrap() + .unwrap_err(); + assert_eq!(error, "provider failure"); + assert!(!cached + .recv_timeout(Duration::from_secs(2)) + .unwrap() + .response_accepted()); + } } diff --git a/src/apps/desktop/src/computer_use/windows_ax_ui.rs b/src/apps/desktop/src/computer_use/windows_ax_ui.rs index 54484c3248..cf8591d7aa 100644 --- a/src/apps/desktop/src/computer_use/windows_ax_ui.rs +++ b/src/apps/desktop/src/computer_use/windows_ax_ui.rs @@ -1123,6 +1123,7 @@ struct ObservedWindow { scope: openbitfun_agent_tools::computer_use_control::ObservationScope, pid: u32, nodes: Vec, + response: std::sync::Arc, } struct WorkerState { initialized: bool, @@ -1311,11 +1312,19 @@ pub(super) fn get_app_state_snapshot_for_window( checkpoint("publish_observation")?; super::control_session::validate_observation_scope(&scope) .map_err(OpenBitFunError::tool)?; + let response = CURRENT_REQUEST + .with_borrow(|request| request.clone()) + .ok_or_else(|| { + OpenBitFunError::tool( + "[AX_WORKER_REQUIRED] Observation requires the provider worker", + ) + })?; state.observed = Some(ObservedWindow { hwnd, scope, pid: window_pid_for(handle).unwrap_or(0), nodes, + response, }); Ok(snapshot) }) @@ -1590,6 +1599,11 @@ fn apply_semantic( ScrollAmount_SmallIncrement, }; super::control_session::capture_allowed().map_err(OpenBitFunError::tool)?; + if !cache.response.response_accepted() { + return Err(OpenBitFunError::tool( + "[AX_OBSERVATION_REQUIRED] The previous observation was not delivered; read the target again before input", + )); + } if cache.hwnd != hwnd || super::control_session::validate_observation_scope(&cache.scope).is_err() || window_pid_for(windows::Win32::Foundation::HWND(hwnd as *mut _)) != Some(cache.pid) From 3db969e1f44ca7abddb98d2f6150d338f7ebd611 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 17 Oct 2026 17:05:14 +0800 Subject: [PATCH 3/4] test(computer-use): run Windows provider regressions independently of Tauri --- .github/workflows/ci.yml | 6 ++++++ src/apps/desktop/src/computer_use/AGENTS.md | 2 ++ 2 files changed, 8 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ee9e62e45d..94d89a3324 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -322,6 +322,12 @@ jobs: if: runner.os != 'Windows' run: cargo test --locked -p openbitfun-desktop --lib + # Exercise production UIA/control code even if the Tauri test executable + # cannot load. Interactive delivery fixtures remain explicitly ignored. + - name: Run Windows Computer Use provider regressions + if: runner.os == 'Windows' + run: node scripts/test-windows-computer-use-native.mjs + # Tauri issue #13419 prevents some Windows MSVC lib-test harnesses from # entering Rust's test runner at all. Keep probing on every run so the # full suite resumes automatically when upstream fixes it. Only the exact diff --git a/src/apps/desktop/src/computer_use/AGENTS.md b/src/apps/desktop/src/computer_use/AGENTS.md index 0f8a9200ed..3a77a87525 100644 --- a/src/apps/desktop/src/computer_use/AGENTS.md +++ b/src/apps/desktop/src/computer_use/AGENTS.md @@ -208,6 +208,8 @@ cargo build -p openbitfun-desktop ``` Windows-only paths (`windows_wgc_capture`, UIA) compile on CI (`windows-latest`). +The Windows job also runs the isolated native-module unit suite independently +of the Tauri test loader; interactive native delivery fixtures remain ignored. UIA observation, semantic input, locate, hit-test and shortcut discovery share one MTA worker. `provider_worker.rs` bounds admission and caller waits: a request has a ten-second deadline and polls control cancellation every 50 ms. UIA also From 731d1e9092ed3e8565344ba1a8d92a6128b145f2 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 17 Oct 2026 17:15:17 +0800 Subject: [PATCH 4/4] fix(computer-use): release idle UIA caches without blocking Stop --- docs/architecture/computer-use-control.md | 4 +- .../src/computer_use/provider_worker.rs | 106 ++++++++++++++++-- .../desktop/src/computer_use/windows_ax_ui.rs | 10 +- 3 files changed, 110 insertions(+), 10 deletions(-) diff --git a/docs/architecture/computer-use-control.md b/docs/architecture/computer-use-control.md index 51059ce225..10f4f45daa 100644 --- a/docs/architecture/computer-use-control.md +++ b/docs/architecture/computer-use-control.md @@ -545,7 +545,9 @@ that worker. Recheck the request fence immediately before input so a late preflight cannot dispatch after abandonment. Once input enters a provider, a timeout, cancellation or provider error means `INPUT_OUTCOME_UNKNOWN`; never replay it or fall back to a coordinate click. Discard abandoned observations on -the owning apartment. A retained observation also carries caller acceptance: +the owning apartment. Idle cache cleanup runs asynchronously there; a new request +waits for that cleanup within its deadline, without queuing input or blocking Stop. +A retained observation also carries caller acceptance: publishing native cache entries before a rejected reply does not make their indices valid for subsequent input. Native input already submitted can still finish after Stop, and a permanently hung provider requires recovery or a host restart. diff --git a/src/apps/desktop/src/computer_use/provider_worker.rs b/src/apps/desktop/src/computer_use/provider_worker.rs index 7c9655c5b9..2be5361a70 100644 --- a/src/apps/desktop/src/computer_use/provider_worker.rs +++ b/src/apps/desktop/src/computer_use/provider_worker.rs @@ -15,6 +15,9 @@ const INPUT_SUBMITTED: u8 = 1; const ABANDONED: u8 = 2; const INPUT_UNKNOWN: u8 = 3; const CANCEL_POLL: Duration = Duration::from_millis(50); +const IDLE: u8 = 0; +const RUNNING: u8 = 1; +const CLEANING: u8 = 2; pub(super) struct Request { operation: &'static str, @@ -124,10 +127,10 @@ impl Request { } } -struct BusyPermit(Arc); +struct BusyPermit(Arc); impl Drop for BusyPermit { fn drop(&mut self) { - self.0.store(false, Ordering::Release); + self.0.store(IDLE, Ordering::Release); } } @@ -135,7 +138,7 @@ type Job = Box; pub(super) struct Worker { sender: SyncSender>, - busy: Arc, + busy: Arc, } impl Worker { @@ -151,10 +154,29 @@ impl Worker { }); Self { sender, - busy: Arc::new(AtomicBool::new(false)), + busy: Arc::new(AtomicU8::new(IDLE)), } } + /// Release thread-affine state promptly when idle. Never queue cleanup + /// behind an occupied provider or make the stopping caller wait for it. + pub(super) fn try_cleanup(&self, cleanup: impl FnOnce(&mut S) + Send + 'static) -> bool { + if self + .busy + .compare_exchange(IDLE, CLEANING, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return false; + } + let permit = BusyPermit(self.busy.clone()); + self.sender + .try_send(Box::new(move |state| { + cleanup(state); + drop(permit); + })) + .is_ok() + } + pub(super) fn run( &self, operation: &'static str, @@ -163,10 +185,22 @@ impl Worker { work: impl FnOnce(&mut S, Arc) -> Result + Send + 'static, ) -> Result, String> { allowed()?; - self.busy.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .map_err(|_| "[AX_PROVIDER_BUSY] The previous native UIA call has not returned. No request was queued and no input was sent; wait for the target application to recover".to_string())?; - let permit = BusyPermit(self.busy.clone()); let request = Arc::new(Request::new(operation, timeout)); + loop { + match self.busy.compare_exchange(IDLE, RUNNING, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => break, + Err(CLEANING) => { + // Stop/start can be adjacent. Await idle COM cleanup within + // this request's deadline instead of reporting a false busy + // provider or queuing input behind an unbounded release. + request.checkpoint("cleanup")?; + if let Err(reason) = allowed() { return Err(request.abandon(&reason)); } + std::thread::sleep(CANCEL_POLL.min(request.deadline.saturating_duration_since(Instant::now()))); + } + Err(_) => return Err("[AX_PROVIDER_BUSY] The previous native UIA call has not returned. No request was queued and no input was sent; wait for the target application to recover".into()), + } + } + let permit = BusyPermit(self.busy.clone()); let worker_request = request.clone(); let (sender, receiver) = mpsc::sync_channel(1); self.sender @@ -243,12 +277,13 @@ mod tests { |_, _| Ok::<_, String>(()), ); assert!(busy.unwrap_err().contains("AX_PROVIDER_BUSY")); + assert!(!worker.try_cleanup(|state| *state = 99)); release.send(()).unwrap(); completed.recv_timeout(Duration::from_secs(2)).unwrap(); // Only the real provider completion releases admission. No replacement // worker or queued action is allowed while the caller is timed out. let start = Instant::now(); - while worker.busy.load(Ordering::Acquire) { + while worker.busy.load(Ordering::Acquire) != IDLE { assert!(start.elapsed() < Duration::from_secs(2)); std::thread::yield_now(); } @@ -426,4 +461,59 @@ mod tests { .unwrap() .response_accepted()); } + + #[test] + fn idle_cleanup_releases_native_state_on_its_owning_thread_without_another_request() { + struct NativeState { + owner: std::thread::ThreadId, + _not_send: std::rc::Rc<()>, + dropped: SyncSender, + } + impl Drop for NativeState { + fn drop(&mut self) { + self.dropped + .send(self.owner == std::thread::current().id()) + .unwrap(); + } + } + let (dropped, receipt) = mpsc::sync_channel(1); + let worker = Worker::new(move || { + Some(NativeState { + owner: std::thread::current().id(), + _not_send: std::rc::Rc::new(()), + dropped, + }) + }); + assert!(worker.try_cleanup(|state| *state = None)); + assert!(receipt.recv_timeout(Duration::from_secs(2)).unwrap()); + } + + #[test] + fn a_new_request_waits_for_idle_cleanup_without_a_spurious_busy_error() { + let worker = Arc::new(Worker::new(|| 0usize)); + let (release, blocked) = mpsc::channel(); + assert!(worker.try_cleanup(move |state| { + blocked.recv().unwrap(); + *state = 42; + })); + let (waiting, ready) = mpsc::sync_channel(1); + let caller = std::thread::spawn(move || { + let mut checks = 0; + worker.run( + "observe", + Duration::from_secs(5), + || { + checks += 1; + if checks == 2 { + waiting.send(()).unwrap(); + } + Ok(()) + }, + |state, _| Ok::<_, String>(*state), + ) + }); + ready.recv_timeout(Duration::from_secs(2)).unwrap(); + release.send(()).unwrap(); + assert_eq!(caller.join().unwrap().unwrap().unwrap(), 42); + } } diff --git a/src/apps/desktop/src/computer_use/windows_ax_ui.rs b/src/apps/desktop/src/computer_use/windows_ax_ui.rs index cf8591d7aa..cb850002ea 100644 --- a/src/apps/desktop/src/computer_use/windows_ax_ui.rs +++ b/src/apps/desktop/src/computer_use/windows_ax_ui.rs @@ -1151,8 +1151,16 @@ thread_local! { pub(super) fn clear_observations() { // Never enqueue behind a stuck provider, and never release COM on the - // cancelling thread. The worker discards it before accepting further work. + // cancelling thread. Idle cleanup releases cached references promptly; + // occupied workers consume the flag when their current call returns. CLEAR_OBSERVATIONS.store(true, std::sync::atomic::Ordering::Release); + if let Some(worker) = UIA_WORKER.get() { + worker.try_cleanup(|state| { + if CLEAR_OBSERVATIONS.swap(false, std::sync::atomic::Ordering::AcqRel) { + state.observed = None; + } + }); + } } pub(super) fn checkpoint(phase: &'static str) -> OpenBitFunResult<()> {