diff --git a/Cargo.lock b/Cargo.lock index 4229815ecb..3d352e59f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1717,7 +1717,7 @@ dependencies = [ [[package]] name = "clipboard-rs" version = "0.3.5" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "clipboard-win", "image", @@ -2438,7 +2438,7 @@ dependencies = [ [[package]] name = "cua-cursor-motion" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "serde", ] @@ -2446,7 +2446,7 @@ dependencies = [ [[package]] name = "cua-driver-contract" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "schemars 1.2.3", "serde", @@ -2457,7 +2457,7 @@ dependencies = [ [[package]] name = "cua-driver-core" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "anyhow", "async-trait", @@ -2497,7 +2497,7 @@ dependencies = [ [[package]] name = "cua-driver-sdk" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "async-trait", "core-foundation 0.10.1", @@ -2526,7 +2526,7 @@ checksum = "f27ae1dd37df86211c42e150270f82743308803d90a6f6e6651cd730d5e1732f" [[package]] name = "cursor-overlay" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "anyhow", "cua-cursor-motion", @@ -11220,7 +11220,7 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" [[package]] name = "pip-preview" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "serde_json", ] @@ -11281,7 +11281,7 @@ checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" [[package]] name = "platform-linux" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "anyhow", "ashpd", @@ -11290,7 +11290,7 @@ dependencies = [ "atspi-connection", "base64 0.22.1", "bitflags 2.13.2", - "clipboard-rs 0.3.5 (git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4)", + "clipboard-rs 0.3.5 (git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451)", "crossbeam-channel", "cua-driver-contract", "cua-driver-core", @@ -11328,7 +11328,7 @@ dependencies = [ [[package]] name = "platform-macos" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "anyhow", "async-trait", @@ -11368,7 +11368,7 @@ dependencies = [ [[package]] name = "platform-windows" version = "0.38.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "anyhow", "async-trait", @@ -17753,7 +17753,7 @@ dependencies = [ [[package]] name = "zbus" version = "5.19.0" -source = "git+https://github.com/maka-agent/cua?rev=cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4#cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" +source = "git+https://github.com/maka-agent/cua?rev=7dfc429ba3db0568c46c49f633b724750f34f451#7dfc429ba3db0568c46c49f633b724750f34f451" dependencies = [ "async-broadcast", "async-executor", diff --git a/Cargo.toml b/Cargo.toml index cd4ba81846..db161d97ac 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -127,8 +127,8 @@ maka-agent-authority = { path = "plugins/agent/authority" } maka-web-api = { path = "plugins/workspace/workspace-web/api" } maka-agent-api = { path = "plugins/agent/api" } maka-transport = { path = "platform/transport" } -cua-driver-sdk = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" } -cua-driver-contract = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" } +cua-driver-sdk = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451" } +cua-driver-contract = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451" } agent-client-protocol = { version = "3.5.0", default-features = false, features = ["unstable_protocol_v2"] } apply-patch = { path = "plugins/workspace/workspace-files/patch" } cap-fs-ext = "4.0.3" @@ -256,7 +256,7 @@ opt-level = 3 opt-level = 3 [patch.crates-io] -zbus = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4" } +zbus = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451" } nom = { git = "https://github.com/maka-agent/cua", rev = "23221967ce6044c1fed303701f43fa54564f0f53" } reqwest = { path = "third-party/reqwest" } gix-index = { path = "third-party/gix-index" } diff --git a/platform/computer-use/Cargo.toml b/platform/computer-use/Cargo.toml index c685b7b953..baf81e31ec 100644 --- a/platform/computer-use/Cargo.toml +++ b/platform/computer-use/Cargo.toml @@ -63,14 +63,14 @@ maka-plugin-core.workspace = true futures-util.workspace = true cua-driver-contract = { workspace = true, optional = true } cua-driver-sdk = { workspace = true, optional = true } -cua-driver-core = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4", optional = true } +cua-driver-core = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451", optional = true } serde.workspace = true schemars.workspace = true uuid.workspace = true libc = { workspace = true, optional = true } reqwest.workspace = true base64 = { version = "0.23.1", optional = true } -cursor-overlay = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4", features = ["theme-authoring"], optional = true } +cursor-overlay = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451", features = ["theme-authoring"], optional = true } sha2 = { workspace = true, optional = true } tempfile = { workspace = true, optional = true } serde_json.workspace = true @@ -92,12 +92,12 @@ tiny-skia = { version = "0.12.0", features = ["png-format"] } cua-skia = { package = "tiny-skia", version = "0.11.4", features = ["png-format"] } [target.'cfg(windows)'.dependencies] -platform-windows = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4", optional = true } +platform-windows = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451", optional = true } windows-sys = { version = "0.61.2", features = ["Win32_Foundation", "Win32_System_Threading", "Win32_Storage_FileSystem"], optional = true } [target.'cfg(target_os = "linux")'.dependencies] image = { version = "0.25.10", default-features = false, features = ["png", "jpeg"], optional = true } -platform-linux = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4", optional = true } +platform-linux = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451", optional = true } [target.'cfg(target_os = "linux")'.dev-dependencies] x11rb = "0.13" @@ -107,7 +107,7 @@ imagesize.workspace = true [target.'cfg(target_os = "macos")'.dependencies] unicode-segmentation = { workspace = true, optional = true } -platform-macos = { git = "https://github.com/maka-agent/cua", rev = "cda1e5c1c7375d7e428e17f3abc75c8cb4d97ae4", optional = true } +platform-macos = { git = "https://github.com/maka-agent/cua", rev = "7dfc429ba3db0568c46c49f633b724750f34f451", optional = true } core-foundation = { version = "0.10.1", optional = true } objc2 = { version = "0.6.5", optional = true } objc2-foundation = { version = "0.3.2", default-features = false, features = ["std", "NSArray", "NSData", "NSString"], optional = true } diff --git a/platform/computer-use/src/api.md b/platform/computer-use/src/api.md index ce7f5f1d03..751c72421c 100644 --- a/platform/computer-use/src/api.md +++ b/platform/computer-use/src/api.md @@ -152,7 +152,7 @@ characters are literal. The default `waitUntil` is `commit`; `load`, `domcontentloaded` and `networkidle` wait for that state after a match. `tab.playwright.waitForLoadState({state?, timeoutMs?})` defaults to `load` and also accepts `domcontentloaded` and `networkidle`. These condition waits default to -3000 ms and cap numeric timeouts at 3000 ms. Zero leaves no event-wait budget: +3000 ms and honor longer explicit timeouts. Zero leaves no event-wait budget: a locator still gets its first probe, current URL/load state may already match, and networkidle times out. `tab.playwright.waitForTimeout(ms)` accepts a nonnegative integer and waits for a fixed duration. For an action-triggered wait, start the @@ -173,8 +173,7 @@ Click options accept `button: 'left'|'right'|'middle'`, `modifiers`, `force` and `timeoutMs`; check options accept `force` and `timeoutMs`. Modifiers are `Alt`, `Control`, `ControlOrMeta`, `Meta` and `Shift`; `ControlOrMeta` uses the selected browser's platform. Other input options accept `timeoutMs`. Timeouts are positive -integers; preparation defaults to 3000 ms and is capped there, or at 5000 ms for -sequential input. An option selection accepts a value string or a descriptor with +integers; preparation defaults to 3000 ms and honors longer explicit timeouts. An option selection accepts a value string or a descriptor with `value`, `label` or a nonnegative integer `index`; arrays select multiple options. Locator mutations use a unique match or the sole visible match. Ambiguous matches @@ -352,7 +351,7 @@ ChatGPT and other third-party adapters remain best effort. - Evaluation exposes read-only DOM, CSS, geometry, range and selection views. It does not expose networking, cookies, module loading or DOM edits. Scrolling, local range boundaries and function-only timers remain supported operations. - The timeout defaults to 3000 ms and numeric values are clamped to 0–3000 ms. + The timeout defaults to 3000 ms; explicit values set the requested duration. A stopped or interrupted script may already have run; it is not automatically executed again. Existing site permissions and tab ownership still apply. @@ -453,7 +452,7 @@ chooser.isMultiple(); await chooser.setFiles(['workspace-file.bin']); ``` -Start the wait before the action that opens the chooser. The default and maximum chooser event wait are 3000 ms; an explicit timeout must be a positive integer. `isMultiple()` is synchronous. `setFiles(pathOrPaths, {timeoutMs})` accepts one path or a nonempty array; a single-file chooser rejects multiple paths. Its optional timeout is validated without discarding an in-flight selection. +Start the wait before the action that opens the chooser. The default chooser event wait is 3000 ms; an explicit timeout must be a positive integer. `isMultiple()` is synchronous. `setFiles(pathOrPaths, {timeoutMs})` accepts one path or a nonempty array; a single-file chooser rejects multiple paths. Its optional timeout is validated without discarding an in-flight selection. Paths refer to the Agent Host's filesystem and remain subject to the caller's workspace file policy. Page access, upload consent for the current HTTP(S) origin, and file-read permission are separate. The supported input snapshot is a regular file. Files are transported as immutable bytes and materialized by the selected Maka browser host; do not translate paths between operating systems yourself. diff --git a/platform/computer-use/src/api.zh-CN.md b/platform/computer-use/src/api.zh-CN.md index 80e83c46fe..c1dd90915a 100644 --- a/platform/computer-use/src/api.zh-CN.md +++ b/platform/computer-use/src/api.zh-CN.md @@ -92,8 +92,7 @@ Turn 清理移除页面桥接并保留数据；重置会创建新的空上下文 `*` 不跨越 `/`，`**` 可以跨越 `/`，其余字符按字面值匹配。`waitUntil` 默认为 `commit`；可指定 `load`、`domcontentloaded` 或 `networkidle`，在 URL 匹配后继续等待。 `tab.playwright.waitForLoadState({state?, timeoutMs?})` 默认等待 `load`，也支持 -`domcontentloaded` 和 `networkidle`。条件等待默认 3000 ms，数值超时上限为 -3000 ms。零不提供事件等待时间：定位器仍检查一次，当前 URL/加载状态可能已匹配， +`domcontentloaded` 和 `networkidle`。条件等待默认 3000 ms，也可以指定更长的等待时间。零不提供事件等待时间：定位器仍检查一次，当前 URL/加载状态可能已匹配， networkidle 则超时。 `tab.playwright.waitForTimeout(ms)` 接受非负整数并等待固定时长。等待某个动作触发的 变化时，先启动等待，再执行动作，最后等待两个 promise。Stop 会取消原来的等待。 @@ -112,8 +111,7 @@ networkidle 则超时。 点击选项为 `button: 'left'|'right'|'middle'`、`modifiers`、`force`、`timeoutMs`； 勾选选项为 `force`、`timeoutMs`。修饰键为 `Alt`、`Control`、`ControlOrMeta`、 `Meta`、`Shift`；`ControlOrMeta` 按所选浏览器的平台解析。其他输入选项为 -`timeoutMs`，它必须是正整数；准备阶段默认且最多 3000 毫秒，逐字符输入最多 -5000 毫秒。选项选择接受 value 字符串，或带 `value`、`label`、非负整数 +`timeoutMs`，它必须是正整数；准备阶段默认等待 3000 毫秒，也可以指定更长的等待时间。选项选择接受 value 字符串，或带 `value`、`label`、非负整数 `index` 的描述对象；数组用于选择多个选项。 定位器选择唯一匹配，或多个匹配中唯一可见的元素；歧义会报错。 @@ -173,7 +171,7 @@ REPL 被取消、超时或耗尽堆内存后，需要 `cua_reset` 并重新绑 - `await browser.user.getTabContext(tabOrId)` 读取当前可见用户标签页的内容，不认领或激活标签页。Host 对当前实际 URL 授权，调用者不能指定替代 URL。隐私标签页和单次读取前后发生 URL 变化的页面会被拒绝。结果为 `{kind:'text',text,title,url,truncated}`，或 `{kind:'document',data,fileName,mimeType,title,url}`；`data` 为实际导出文件的 `Uint8Array`。Host 提供文件发布服务时，文档还包含本 Session 持有的真实 `path`。Google Workspace 可提供 Markdown、PDF 或 XLSX，普通页面可包含当前选择，YouTube 视频页可包含字幕。`maka` 实现固定读取，`chatgpt` 在支持时复用原版读取协议；第三方插件按实际能力尽力支持。 - `await tab.playwright.evaluate(pageFunction, arg?, {timeoutMs}?)` 在当前页面执行自包含函数；字符串按表达式处理。`locator.evaluate` 向函数传入 `(element, arg)`，`locator.evaluateAll` 传入 `(elements, arg)`；单元素形式要求目标明确，集合形式允许零匹配。参数须可 JSON 序列化。结果为普通数据：忽略函数和 Symbol，BigInt 转为字符串，非有限数字转为 `null`。本 API 的结果上限为深度 8、数组 2000 项、对象 200 个键、字符串 200000 个 UTF-16 单元，按需使用 `[MaxDepth]`、`[Circular]` 和 `[Truncated]` 标记。 -- 执行环境提供 DOM、CSS、几何、Range 和 Selection 的只读视图，不开放网络、Cookie、模块加载或 DOM 修改。滚动、局部 Range 边界和仅接收函数的定时器仍受支持。超时默认为 3000 ms，数字值限制在 0–3000 ms。停止或中断时，脚本可能已经执行；系统不会自动重放。现有站点授权和标签页归属继续生效。 +- 执行环境提供 DOM、CSS、几何、Range 和 Selection 的只读视图，不开放网络、Cookie、模块加载或 DOM 修改。滚动、局部 Range 边界和仅接收函数的定时器仍受支持。超时默认为 3000 ms，也可以指定更长的等待时间。停止或中断时，脚本可能已经执行；系统不会自动重放。现有站点授权和标签页归属继续生效。 浏览器指针方法包括 `tab.cua.move({x,y,keys?})`、 `click({x,y,button?,keypress?})`、`double_click({x,y,keypress?})`、 @@ -267,7 +265,7 @@ chooser.isMultiple(); await chooser.setFiles(['workspace-file.bin']); ``` -先开始等待，再执行打开文件选择器的操作。默认及最长事件等待时间为 3000 毫秒；显式 timeout 必须是正整数。`isMultiple()` 同步返回。`setFiles(pathOrPaths, {timeoutMs})` 接受单一路径或非空数组；单选选择器拒绝多个路径。可选 timeout 会校验，但不会丢弃已经派发的文件选择请求。 +先开始等待，再执行打开文件选择器的操作。默认事件等待时间为 3000 毫秒；显式 timeout 必须是正整数。`isMultiple()` 同步返回。`setFiles(pathOrPaths, {timeoutMs})` 接受单一路径或非空数组；单选选择器拒绝多个路径。可选 timeout 会校验，但不会丢弃已经派发的文件选择请求。 路径属于 Agent Host 的文件系统，仍受调用者的工作区文件策略约束。网页访问、当前 HTTP(S) 来源的上传批准、文件读取权限互相独立。当前输入快照支持普通文件。文件作为不可变字节传输，由所选 Maka 浏览器宿主物化；不要自行换算操作系统之间的文件路径。 diff --git a/platform/computer-use/src/api.zh-TW.md b/platform/computer-use/src/api.zh-TW.md index 8620ce3e61..b9a83b1666 100644 --- a/platform/computer-use/src/api.zh-TW.md +++ b/platform/computer-use/src/api.zh-TW.md @@ -92,8 +92,7 @@ Turn 清理移除頁面橋接並保留資料；重設會建立新的空上下文 `*` 不跨越 `/`，`**` 可以跨越 `/`，其餘字元依字面值比對。`waitUntil` 預設為 `commit`；可指定 `load`、`domcontentloaded` 或 `networkidle`，在 URL 相符後繼續等待。 `tab.playwright.waitForLoadState({state?, timeoutMs?})` 預設等待 `load`，亦支援 -`domcontentloaded` 和 `networkidle`。條件等待預設 3000 ms，數值逾時上限為 -3000 ms。零不提供事件等待時間：定位器仍檢查一次，目前 URL/載入狀態可能已符合， +`domcontentloaded` 和 `networkidle`。條件等待預設 3000 ms，也可以指定更長的等待時間。零不提供事件等待時間：定位器仍檢查一次，目前 URL/載入狀態可能已符合， networkidle 則逾時。 `tab.playwright.waitForTimeout(ms)` 接受非負整數並等待固定時長。等待某個動作觸發的 變化時，先啟動等待，再執行動作，最後等待兩個 promise。Stop 會取消原來的等待。 @@ -112,8 +111,7 @@ networkidle 則逾時。 點擊選項為 `button: 'left'|'right'|'middle'`、`modifiers`、`force`、`timeoutMs`； 勾選選項為 `force`、`timeoutMs`。修飾鍵為 `Alt`、`Control`、`ControlOrMeta`、 `Meta`、`Shift`；`ControlOrMeta` 按所選瀏覽器的平台解析。其他輸入選項為 -`timeoutMs`，必須是正整數；準備階段預設且最多 3000 毫秒，逐字元輸入最多 -5000 毫秒。選項選擇接受 value 字串，或帶 `value`、`label`、非負整數 +`timeoutMs`，必須是正整數；準備階段預設等待 3000 毫秒，也可以指定更長的等待時間。選項選擇接受 value 字串，或帶 `value`、`label`、非負整數 `index` 的描述物件；陣列用來選擇多個選項。 定位器選擇唯一匹配，或多個匹配中唯一可見的元素；有歧義時會報錯。 @@ -173,7 +171,7 @@ REPL 被取消、逾時或耗盡堆積記憶體後，需要 `cua_reset` 並重 - `await browser.user.getTabContext(tabOrId)` 讀取目前可見使用者分頁的內容，不認領或啟用分頁。Host 對目前實際 URL 授權，呼叫者不能指定替代 URL。隱私分頁和單次讀取前後發生 URL 變更的頁面會被拒絕。結果為 `{kind:'text',text,title,url,truncated}`，或 `{kind:'document',data,fileName,mimeType,title,url}`；`data` 為實際匯出檔案的 `Uint8Array`。Host 提供檔案發布服務時，文件還包含本 Session 持有的真實 `path`。Google Workspace 可提供 Markdown、PDF 或 XLSX，一般頁面可包含目前選取內容，YouTube 影片頁可包含字幕。`maka` 實作固定讀取，`chatgpt` 在支援時重用原版讀取協定；第三方擴充功能依實際能力盡力支援。 - `await tab.playwright.evaluate(pageFunction, arg?, {timeoutMs}?)` 在目前頁面執行自包含函式；字串按運算式處理。`locator.evaluate` 向函式傳入 `(element, arg)`，`locator.evaluateAll` 傳入 `(elements, arg)`；單元素形式要求目標明確，集合形式允許零匹配。參數須可 JSON 序列化。結果為一般資料：略過函式和 Symbol，BigInt 轉為字串，非有限數字轉為 `null`。本 API 的結果上限為深度 8、陣列 2000 項、物件 200 個鍵、字串 200000 個 UTF-16 單元，視需要使用 `[MaxDepth]`、`[Circular]` 和 `[Truncated]` 標記。 -- 執行環境提供 DOM、CSS、幾何、Range 和 Selection 的唯讀檢視，不開放網路、Cookie、模組載入或 DOM 修改。捲動、局部 Range 邊界和僅接收函式的計時器仍受支援。逾時預設為 3000 ms，數字值限制在 0–3000 ms。停止或中斷時，指令碼可能已經執行；系統不會自動重播。現有網站授權和分頁歸屬繼續生效。 +- 執行環境提供 DOM、CSS、幾何、Range 和 Selection 的唯讀檢視，不開放網路、Cookie、模組載入或 DOM 修改。捲動、局部 Range 邊界和僅接收函式的計時器仍受支援。逾時預設為 3000 ms，也可以指定更長的等待時間。停止或中斷時，指令碼可能已經執行；系統不會自動重播。現有網站授權和分頁歸屬繼續生效。 瀏覽器指標方法包括 `tab.cua.move({x,y,keys?})`、 `click({x,y,button?,keypress?})`、`double_click({x,y,keypress?})`、 @@ -267,7 +265,7 @@ chooser.isMultiple(); await chooser.setFiles(['workspace-file.bin']); ``` -先開始等待，再執行開啟檔案選擇器的操作。預設及最長事件等待時間為 3000 毫秒；明確指定的 timeout 必須是正整數。`isMultiple()` 同步回傳。`setFiles(pathOrPaths, {timeoutMs})` 接受單一路徑或非空陣列；單選選擇器拒絕多個路徑。可選 timeout 會驗證，但不會丟棄已派發的檔案選擇請求。 +先開始等待，再執行開啟檔案選擇器的操作。預設事件等待時間為 3000 毫秒；明確指定的 timeout 必須是正整數。`isMultiple()` 同步回傳。`setFiles(pathOrPaths, {timeoutMs})` 接受單一路徑或非空陣列；單選選擇器拒絕多個路徑。可選 timeout 會驗證，但不會丟棄已派發的檔案選擇請求。 路徑屬於 Agent Host 的檔案系統，仍受呼叫者的工作區檔案原則約束。網頁存取、目前 HTTP(S) 來源的上傳批准、檔案讀取權限彼此獨立。目前輸入快照支援一般檔案。檔案以不可變位元組傳輸，由所選 Maka 瀏覽器宿主實體化；不要自行換算作業系統之間的檔案路徑。 diff --git a/platform/computer-use/src/browser.rs b/platform/computer-use/src/browser.rs index 1d4d404ad1..6b632f991b 100644 --- a/platform/computer-use/src/browser.rs +++ b/platform/computer-use/src/browser.rs @@ -168,10 +168,10 @@ struct ReadBudget { remaining: Duration, } impl ReadBudget { - fn new(options: &BrowserReadOptions) -> Self { - Self { - remaining: Duration::from_millis(options.timeout_ms.unwrap_or(3000).min(3000)), - } + fn new(options: &BrowserReadOptions) -> Result { + Ok(Self { + remaining: crate::protocol::timeout_duration(options.timeout_ms.unwrap_or(3000) as f64)?, + }) } } #[derive(Clone, Copy)] @@ -1197,7 +1197,7 @@ impl Browsers { self.use_tab(&identity, turn, cancellation).await?; let tab = self.targets.get_mut(&handle).expect("checked browser handle"); tab.last_turn = turn.to_owned(); - let mut budget = Some(ReadBudget { remaining:options.budget() }); + let mut budget = Some(ReadBudget { remaining:options.budget()? }); tab.authorize_with_budget(None,invocation,cancellation,language, extension::ClaimIntent::Implicit,&mut budget).await?; tab.playwright_evaluate(locator.as_ref(),mode,&script_json,&options, @@ -1220,7 +1220,7 @@ impl Browsers { self.use_tab(&identity, turn, cancellation).await?; let tab = self.targets.get_mut(&handle).unwrap(); tab.last_turn = turn.to_owned(); - let mut budget = Some(ReadBudget::new(&options)); + let mut budget = Some(ReadBudget::new(&options)?); tab.authorize_with_budget( None, invocation, @@ -1966,6 +1966,8 @@ impl Browsers { tab.frame = None; tab.previous = None; } + extension::drain_extension_cleanup(&self.host().await, self.owner, Some(turn), None, None) + .await?; let finished = self .host() .await @@ -2045,6 +2047,7 @@ impl Browsers { failures.push(error.to_string()); } let host = self.host().await; + extension::drain_extension_cleanup(&host, self.owner, None, None, None).await?; let mut host = host.lock().await; let cursor_result = host.finalize_cursors(self.owner, cursor).await; let tabs_result = host.finish(self.owner, None).await; @@ -3797,11 +3800,11 @@ mod tests { } #[tokio::test] - async fn healthy_browser_discovery_and_opencli_preference_keep_other_provider_uncertainty() { + async fn healthy_browser_discovery_keeps_other_provider_uncertainty() { for unhealthy_first in [true, false] { let (healthy, _, healthy_server) = inventory_provider("actual-cdp", 3, vec![]).await; let (unhealthy, generation, unhealthy_server) = - inventory_provider("unconfirmed-cdp", 3, vec![]).await; + inventory_provider("unconfirmed-cdp", 2, vec![]).await; let mut browsers = Browsers::default(); let connections = if unhealthy_first { vec![unhealthy.clone(), healthy] @@ -3850,7 +3853,7 @@ mod tests { .invoke( Command::GetBrowser { options: BrowserOptions { - id: Some("opencli".into()), + id: Some("actual-cdp".into()), url: None, extension_instance_id: None, }, @@ -3870,7 +3873,7 @@ mod tests { 1 ); assert_eq!(healthy_server.await.unwrap().len(), 6); - assert_eq!(unhealthy_server.await.unwrap().len(), 6); + assert_eq!(unhealthy_server.await.unwrap().len(), 4); } } @@ -4001,9 +4004,12 @@ mod tests { }; let tab = browsers.targets.get_mut(&handle).unwrap(); assert!(tab.socket.is_none(), "binding must remain lazy"); - let mut budget = Some(ReadBudget::new(&BrowserReadOptions { - timeout_ms: Some(if cancel_at_handshake { 3000 } else { 200 }), - })); + let mut budget = Some( + ReadBudget::new(&BrowserReadOptions { + timeout_ms: Some(if cancel_at_handshake { 3000 } else { 200 }), + }) + .unwrap(), + ); // This is the same bounded connect used by metadata preparation; // inventory and consent have their own behavior tests. let result = tokio::time::timeout( diff --git a/platform/computer-use/src/browser/downloads.rs b/platform/computer-use/src/browser/downloads.rs index 67f4d96980..50b403a982 100644 --- a/platform/computer-use/src/browser/downloads.rs +++ b/platform/computer-use/src/browser/downloads.rs @@ -98,18 +98,14 @@ impl Store { } } -pub(super) fn timeout(options: &BrowserDownloadOptions) -> Duration { - let millis = options - .timeout_ms - .as_ref() - .and_then(Value::as_f64) - .unwrap_or(3000.0) - .clamp(0.0, 120_000.0); - Duration::from_secs_f64(millis / 1000.0) -} - -pub(super) fn locator_timeout(options: &BrowserDownloadOptions) -> Duration { - timeout(options).min(Duration::from_millis(3000)) +pub(super) fn timeout(options: &BrowserDownloadOptions) -> Result { + crate::protocol::timeout_duration( + options + .timeout_ms + .as_ref() + .and_then(Value::as_f64) + .unwrap_or(3000.0), + ) } fn cleanup(error: impl std::fmt::Display) -> ToolError { @@ -298,7 +294,7 @@ impl Tab { session_metadata, authorization, cancellation.clone(), - timeout(&options), + timeout(&options)?, ) .await } diff --git a/platform/computer-use/src/browser/downloads/tests.rs b/platform/computer-use/src/browser/downloads/tests.rs index f018c27e26..3529b0ad66 100644 --- a/platform/computer-use/src/browser/downloads/tests.rs +++ b/platform/computer-use/src/browser/downloads/tests.rs @@ -142,22 +142,3 @@ fn attachment_header_is_anchored_and_preserves_extension_header_bytes() { ); } } - -#[test] -fn wait_and_locator_budgets_preserve_the_selected_service_clamps() { - for (value, expected) in [ - (None, 3000), - (Some(json!("ignored")), 3000), - (Some(json!(-1)), 0), - (Some(json!(0)), 0), - (Some(json!(10000)), 10000), - (Some(json!(999999)), 120000), - ] { - let options = BrowserDownloadOptions { timeout_ms: value }; - assert_eq!(timeout(&options), Duration::from_millis(expected)); - assert_eq!( - locator_timeout(&options), - Duration::from_millis(expected.min(3000)) - ); - } -} diff --git a/platform/computer-use/src/browser/extension.rs b/platform/computer-use/src/browser/extension.rs index add3664658..59b956b160 100644 --- a/platform/computer-use/src/browser/extension.rs +++ b/platform/computer-use/src/browser/extension.rs @@ -806,16 +806,8 @@ struct ControlDrain { } impl ControlDrain { async fn wait(self) -> Result<(), ToolError> { - let deadline = tokio::time::Instant::now() + Duration::from_secs(10); for gate in self.gates { - if tokio::time::timeout_at(deadline, gate.idle()) - .await - .is_err() - { - return Err(ToolError::CleanupUnconfirmed( - "extension control futures remain active; original custody retained".to_owned(), - )); - } + gate.idle().await; } Ok(()) } @@ -2912,6 +2904,19 @@ impl Host { "extension completion belongs to another physical generation", )); } + if completion + .wire_context + .as_ref() + .is_some_and(|context| context.host_session_id() != owner.to_string()) + || completion.host_operation.as_ref().is_some_and(|context| { + !self + .extension + .get(&(owner, browser.clone())) + .is_some_and(|state| state.calls.contains_key(&context.operation_id)) + }) + { + continue; + } if completion.outcome.is_none() { unsettled = true; continue; @@ -4565,13 +4570,10 @@ impl Host { let caps = extension .capabilities() .map_err(provider::extension_error)?; - let mut events = extension.subscribe_events(); - let deadline = tokio::time::Instant::now() + Duration::from_secs(10); - while self.reconcile_provider(owner, &provider).await? { - match tokio::time::timeout_at(deadline, events.recv()).await { - Ok(Ok(_)) | Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => {} - _ => return Err(unknown("owner teardown retains pending extension custody")), - } + if self.reconcile_provider(owner, &provider).await? { + return Err(unknown( + "extension cleanup requires the original control operations to settle", + )); } if self .uncertain_creations @@ -4647,86 +4649,8 @@ impl Host { } } for identity in identities { - let state = self.tabs[&identity] - .extension - .as_ref() - .ok_or_else(|| unknown("owner teardown lost physical custody"))? - .clone(); - if state.unknown.is_some() || state.cdp_effect_unknown.is_some() { - return Err(unknown("owner teardown retains unknown physical effects")); - } - if state.claim.is_some() - || !matches!(state.fixed_preparation, FixedPreparation::NotAttempted) - { - self.cleanup_extension_control(owner, &identity, FixedCleanupPurpose::Owner) - .await?; - } - if let Some(claim) = &state.claim { - if caps.header_scope == ExtensionHeaderPolicyScope::Claim - && (state.headers_active || state.preparation_effects.headers_may_exist) - { - let result = scope - .dispatch_retirement(scope.control.clear_scoped_headers(claim)) - .await; - self.reconcile_provider(owner, &provider).await?; - if !result.map_err(provider::extension_error)?.cleanup_confirmed { - return Err(unknown("owner scoped headers remain unconfirmed")); - } - } - let drain = scope - .dispatch_retirement( - scope - .control - .wait_claim_cdp_drained(claim, Duration::from_secs(10)), - ) - .await - .map_err(provider::extension_error)?; - self.tabs - .get_mut(&identity) - .and_then(|lease| lease.extension.as_mut()) - .ok_or_else(|| unknown("owner drain lost custody"))? - .cdp_drain = Some(drain); - self.reconcile_provider(owner, &provider).await?; - if !self.tabs[&identity] - .extension - .as_ref() - .is_some_and(|state| state.released) - { - self.begin_extension_cleanup(&identity)?; - let result = scope - .dispatch_retirement(scope.control.detach_tab(claim)) - .await; - self.reconcile_provider(owner, &provider).await?; - result.map_err(provider::extension_error)?; - } - } - self.retire_extension_cleanup(owner, &provider, &identity, &scope.control) + self.park_extension_tab(owner, &provider, &identity, &scope) .await?; - let lease = self - .tabs - .get_mut(&identity) - .ok_or_else(|| unknown("owner retirement lost lease"))?; - let state = lease - .extension - .as_ref() - .ok_or_else(|| unknown("owner retirement lost custody"))?; - if state.claim.is_some() - || !state.fixed_preparation.settled() - || !state.preparation_effects.settled() - { - return Err(unknown("owner physical retirement remains unconfirmed")); - } - // Source cleanup provenance stays with this connection; no - // physical claim, prepared grant or CDP route survives parking. - lease.extension = Some(ClaimState { - creation: state.creation.clone(), - creation_operation: state.creation_operation.clone(), - group: state.group.clone(), - revoked: true, - released: true, - ..Default::default() - }); - lease.disposition = Disposition::Parked; } if let Some(state) = self.extension.get_mut(&(owner, browser)) { for call in state.calls.values() { @@ -4741,6 +4665,92 @@ impl Host { Ok(Vec::new()) } + async fn park_extension_tab( + &mut self, + owner: uuid::Uuid, + provider: &Provider, + identity: &TabIdentity, + scope: &CallScope, + ) -> Result<(), ToolError> { + let caps = scope + .control + .capabilities() + .map_err(provider::extension_error)?; + let state = self.tabs[identity] + .extension + .as_ref() + .ok_or_else(|| unknown("owner teardown lost physical custody"))? + .clone(); + if state.unknown.is_some() || state.cdp_effect_unknown.is_some() { + return Err(unknown("owner teardown retains unknown physical effects")); + } + if state.claim.is_some() + || !matches!(state.fixed_preparation, FixedPreparation::NotAttempted) + { + self.cleanup_extension_control(owner, identity, FixedCleanupPurpose::Owner) + .await?; + } + if let Some(claim) = &state.claim { + if caps.header_scope == ExtensionHeaderPolicyScope::Claim + && (state.headers_active || state.preparation_effects.headers_may_exist) + { + let result = scope + .dispatch_retirement(scope.control.clear_scoped_headers(claim)) + .await; + self.reconcile_provider(owner, provider).await?; + if !result.map_err(provider::extension_error)?.cleanup_confirmed { + return Err(unknown("owner scoped headers remain unconfirmed")); + } + } + if state.cdp_drain.is_none() { + return Err(unknown( + "extension claim must be drained before physical retirement", + )); + } + self.reconcile_provider(owner, provider).await?; + if !self.tabs[identity] + .extension + .as_ref() + .is_some_and(|state| state.released) + { + self.begin_extension_cleanup(identity)?; + let result = scope + .dispatch_retirement(scope.control.detach_tab(claim)) + .await; + self.reconcile_provider(owner, provider).await?; + result.map_err(provider::extension_error)?; + } + } + self.retire_extension_cleanup(owner, provider, identity, &scope.control) + .await?; + let lease = self + .tabs + .get_mut(identity) + .ok_or_else(|| unknown("owner retirement lost lease"))?; + let state = lease + .extension + .as_ref() + .ok_or_else(|| unknown("owner retirement lost custody"))?; + if state.claim.is_some() + || !state.fixed_preparation.settled() + || !state.preparation_effects.settled() + { + return Err(unknown("owner physical retirement remains unconfirmed")); + } + // Source cleanup provenance stays with this connection; no + // physical claim, prepared grant or CDP route survives parking. + lease.extension = Some(ClaimState { + creation: state.creation.clone(), + creation_operation: state.creation_operation.clone(), + group: state.group.clone(), + revoked: true, + released: true, + ..Default::default() + }); + lease.disposition = Disposition::Parked; + Ok(()) + } + pub(super) async fn finish_extension( &mut self, owner: uuid::Uuid, @@ -4783,17 +4793,10 @@ impl Host { .capabilities() .map_err(provider::extension_error)?; let browser = BrowserIdentity::of(&provider, &extension.identity().generation)?; - let mut events = extension.subscribe_events(); - let deadline = tokio::time::Instant::now() + Duration::from_secs(10); - while self.reconcile_provider(owner, &provider).await? { - match tokio::time::timeout_at(deadline, events.recv()).await { - Ok(Ok(_)) | Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => {} - _ => { - return Err(ToolError::CleanupUnconfirmed( - "extension operations remain in flight; custody retained".into(), - )); - } - } + if self.reconcile_provider(owner, &provider).await? { + return Err(unknown( + "extension cleanup requires the original control operations to settle", + )); } let candidates: Vec<_> = self .tabs @@ -4978,20 +4981,12 @@ impl Host { self.cleanup_extension_control(owner, identity, purpose.clone()) .await?; } - if let Some(claim) = &state.claim { - let drain = scope - .dispatch_retirement( - scope - .control - .wait_claim_cdp_drained(claim, Duration::from_secs(10)), - ) - .await - .map_err(provider::extension_error)?; - self.tabs - .get_mut(identity) - .and_then(|lease| lease.extension.as_mut()) - .ok_or_else(|| unknown("extension drain lost its exact claim"))? - .cdp_drain = Some(drain); + if state.claim.is_some() { + if state.cdp_drain.is_none() { + return Err(unknown( + "extension claim must be drained before Turn cleanup", + )); + } if self.reconcile_provider(owner, &provider).await? { return Err(ToolError::CleanupUnconfirmed( "extension CDP settled but retained operation custody is still pending" @@ -5625,11 +5620,7 @@ async fn register( } } Ok(ExtensionProviderEvent::ClaimUnavailable { claim, reason }) => { - let explicit_fixed = provider.extension().is_some_and(|extension| { - extension.capabilities().is_ok_and(|caps| { - caps.session_preparation == ExtensionSessionPreparation::ExplicitFixed - }) - }); + tracing::warn!(%owner, %claim, %reason, "Browser extension claim unavailable"); for lease in ledger.tabs.values_mut().filter(|lease| { lease.creator.owner == owner && lease.creator.provider == provider @@ -5641,33 +5632,10 @@ async fn register( .as_ref() .is_some_and(|owned| owned.claim_id() == claim) { - let logical_retained = explicit_fixed - && state.fixed_preparation.cleanup_receipt().is_some_and( - |receipt| { - receipt.wire_context().host_session_id() - == owner.to_string() - && receipt.generation() == lease.creator.generation - && provider.extension().is_some_and(|extension| { - receipt.generation() - == extension.identity().generation - }) - && state.claim.as_ref().is_some_and(|physical| { - physical.tab_id() == receipt.tab_id() - }) - }, - ); state.revoked = true; - if state.cleanup_started || logical_retained { - // This is a physical-route observation, never - // logical cleanup or underlying-effect proof. - // Core retains every actual pending CDP effect - // independently before broadcasting this event. - // Keep all previous Unknown and physical flags; - // only exact cleanup+drain can retire the claim. - state.cleanup_revocation.get_or_insert(reason.clone()); - } else { - state.unknown = Some(reason.clone()); - } + // Route loss does not establish an unknown effect. + // Pending CDP and preparation custody remains in its own ledger. + state.cleanup_revocation.get_or_insert(reason.clone()); } } } @@ -5752,6 +5720,130 @@ async fn register( Ok(()) } +pub(super) async fn drain_extension_cleanup( + host: &Arc>, + owner: uuid::Uuid, + turn: Option<&str>, + adapter: Option, + target: Option<&TabIdentity>, +) -> Result<(), ToolError> { + let providers: Vec<_> = host + .lock() + .await + .extension + .iter() + .filter(|((id, browser), state)| { + *id == owner + && !state.owner_parked + && adapter.is_none_or(|adapter| state.provider.extension_adapter() == Some(adapter)) + && target.is_none_or(|target| target.browser == *browser) + }) + .map(|(_, state)| state.provider.clone()) + .collect(); + for provider in providers { + let extension = provider + .extension() + .ok_or_else(|| failed("extension provider required"))?; + let mut events = extension.subscribe_events(); + loop { + let pending = host + .lock() + .await + .reconcile_provider(owner, &provider) + .await?; + if !pending { + break; + } + match events.recv().await { + Ok(ExtensionProviderEvent::GenerationClosed { reason }) => { + return Err(unknown(reason)); + } + Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + Err(error) => return Err(unknown(error)), + } + } + let claims = { + let mut ledger = host.lock().await; + let claims: Vec<_> = ledger + .tabs + .iter() + .filter(|(identity, lease)| { + lease.creator.owner == owner + && lease.creator.provider == provider + && !matches!(lease.disposition, Disposition::Parked) + && target.is_none_or(|target| *identity == target) + && turn.is_none_or(|turn| { + !matches!(lease.retention, Retention::Handoff) + && (lease.creator.turn == turn + || lease.extension.as_ref().is_some_and(|state| { + !matches!( + state.fixed_preparation, + FixedPreparation::NotAttempted + ) + })) + }) + }) + .filter_map(|(identity, lease)| { + lease + .extension + .as_ref()? + .claim + .clone() + .map(|claim| (identity.clone(), claim, lease.creator.turn.clone())) + }) + .collect(); + let mut planned = Vec::new(); + for (identity, claim, claim_turn) in claims { + if let Some(turn) = turn + && ledger.fixed_cleanup_protected( + owner, + &identity, + &FixedCleanupPurpose::Turn { + turn: turn.to_owned(), + }, + )? + { + continue; + } + ledger.check_owned_input(&identity)?; + let scope = ledger.extension_context( + owner, + &provider, + context_operation(turn.unwrap_or(&claim_turn)), + crate::desktop::NativeEpochId::new(), + CallTarget::Retirement, + false, + )?; + planned.push((identity, claim, scope)); + } + planned + }; + for (identity, claim, scope) in claims { + let drain = scope + .dispatch_retirement(scope.control.wait_claim_cdp_drained(&claim)) + .await + .map_err(provider::extension_error)?; + let mut ledger = host.lock().await; + let state = ledger + .tabs + .get_mut(&identity) + .filter(|lease| lease.creator.owner == owner && lease.creator.provider == provider) + .and_then(|lease| lease.extension.as_mut()) + .filter(|state| { + state + .claim + .as_ref() + .is_some_and(|current| current.claim_id() == claim.claim_id()) + }) + .ok_or_else(|| { + unknown("extension drain no longer belongs to the retained claim") + })?; + state.cdp_drain = Some(drain); + } + } + Ok(()) +} + impl Browsers { pub(super) async fn drain_handoff_controls(&self) -> Result<(), ToolError> { let drain = self.host().await.lock().await.seal_extension_controls( @@ -6395,6 +6487,14 @@ impl Browsers { let host = self.host().await; let generation = extension.identity().generation.clone(); let identity = BrowserIdentity::of(&provider, &generation)?.tab(&id.to_string()); + let revoked = host.lock().await.tabs.get(&identity).is_some_and(|lease| { + lease.creator.owner == self.owner + && !matches!(lease.disposition, Disposition::Parked) + && lease.extension.as_ref().is_some_and(|state| state.revoked) + }); + if revoked { + drain_extension_cleanup(&host, self.owner, None, None, Some(&identity)).await?; + } let history = self .acquisition_history(&provider) .into_iter() @@ -6414,6 +6514,24 @@ impl Browsers { }, true, )?; + if ledger.tabs.get(&identity).is_some_and(|lease| { + lease.creator.owner == self.owner + && !matches!(lease.disposition, Disposition::Parked) + && lease.extension.as_ref().is_some_and(|state| state.revoked) + }) { + ledger.check_owned_input(&identity)?; + let retirement = ledger.extension_context( + self.owner, + &provider, + context_operation(invocation.turn()), + self.epoch, + CallTarget::Retirement, + false, + )?; + ledger + .park_extension_tab(self.owner, &provider, &identity, &retirement) + .await?; + } if !ledger.begin_handoff_transfer( self.owner, &provider, @@ -8924,11 +9042,7 @@ impl Tab { .await?; } let drain = scope - .dispatch_retirement( - scope - .control - .wait_claim_cdp_drained(&claim, Duration::from_secs(10)), - ) + .dispatch_retirement(scope.control.wait_claim_cdp_drained(&claim)) .await .map_err(provider::extension_error)?; { diff --git a/platform/computer-use/src/browser/extension/inventory_tests.rs b/platform/computer-use/src/browser/extension/inventory_tests.rs index bc35b7de15..f1013b3088 100644 --- a/platform/computer-use/src/browser/extension/inventory_tests.rs +++ b/platform/computer-use/src/browser/extension/inventory_tests.rs @@ -722,3 +722,214 @@ async fn retained_chatgpt_page_keeps_origin_when_a_new_native_generation_claims_ current.assert_no_frame_and_close().await; } } + +#[tokio::test] +async fn revoked_tab_retires_its_old_claim_and_can_be_explicitly_claimed_again() { + let mut fixture = Fixture::open().await; + let provider = fixture.provider(); + let owner = uuid::Uuid::new_v4(); + let identity = browser(&provider).tab("17"); + let host = Arc::new(tokio::sync::Mutex::new(Host::default())); + register(&host, owner, &provider).await.unwrap(); + let epoch = crate::desktop::NativeEpochId::new(); + let operation = context_operation("recover-turn"); + let call = { + let mut ledger = host.lock().await; + ledger + .extension_reserve(owner, &provider, "recover-turn", "17", Origin::AgentCreated) + .unwrap(); + ledger + .extension_context( + owner, + &provider, + operation.clone(), + epoch, + CallTarget::Tab { + origin: Origin::UserClaimed, + tab: Some("17".into()), + }, + true, + ) + .unwrap() + }; + let attach = async { + let frame = read_frame(&mut fixture.server).await; + assert_eq!(frame["request"]["method"], "attach"); + result( + &mut fixture.server, + frame["request"]["id"].as_u64().unwrap(), + json!({"claim":"original-claim","tabId":17,"targetId":"target-17"}), + ) + .await; + }; + let (attached, ()) = + bounded(async { tokio::join!(call.dispatch(call.control.attach_tab(17)), attach) }).await; + attached.unwrap(); + host.lock() + .await + .reconcile_provider(owner, &provider) + .await + .unwrap(); + write_frame(&mut fixture.server, json!({"kind":"controlEvent","event":{"kind":"claimUnavailable","data":{"claim":"original-claim","reason":"fixture route revoked"}}})).await; + bounded(async { + loop { + if host.lock().await.tabs[&identity] + .extension + .as_ref() + .unwrap() + .revoked + { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + { + let mut ledger = host.lock().await; + assert!( + matches!(ledger.extension_reserve(owner, &provider, "recover-turn", "17", Origin::UserClaimed), + Err(ToolError::Failed(message)) if message.contains("explicitly claim")) + ); + let state = ledger.tabs[&identity].extension.as_ref().unwrap(); + assert!( + state.unknown.is_none(), + "route loss must not invent an unknown Chrome effect" + ); + assert_eq!( + state.cleanup_revocation.as_deref(), + Some("fixture route revoked") + ); + } + let retirement = host + .lock() + .await + .extension_context( + owner, + &provider, + context_operation("recover-turn"), + epoch, + CallTarget::Retirement, + false, + ) + .unwrap(); + let retire = async { + drain_extension_cleanup(&host, owner, None, None, Some(&identity)).await?; + host.lock() + .await + .park_extension_tab(owner, &provider, &identity, &retirement) + .await + }; + let peer = async { + let fence = read_frame(&mut fixture.server).await; + assert_eq!(fence["request"]["method"], "fenceCdp"); + host.lock() + .await + .extension_reserve(owner, &provider, "recover-turn", "18", Origin::UserClaimed) + .expect("another page remains usable while the original drain waits"); + result(&mut fixture.server, fence["request"]["id"].as_u64().unwrap(), + json!({"claim":"original-claim","generation":fixture.extension.identity().generation,"fenced":true,"pending":[]})).await; + let detach = read_frame(&mut fixture.server).await; + assert_eq!(detach["request"]["method"], "detach"); + assert_eq!(detach["request"]["claim"], "original-claim"); + result( + &mut fixture.server, + detach["request"]["id"].as_u64().unwrap(), + json!({"detached":true,"proof":"detach_ack"}), + ) + .await; + let retired = read_frame(&mut fixture.server).await; + assert_eq!(retired["request"]["method"], "retireClaim"); + result( + &mut fixture.server, + retired["request"]["id"].as_u64().unwrap(), + json!({"retired":true,"claim":"original-claim"}), + ) + .await; + Ok::<(), ToolError>(()) + }; + bounded(async { tokio::try_join!(retire, peer) }) + .await + .unwrap(); + let operation = context_operation("recover-turn"); + let replacement = { + let mut ledger = host.lock().await; + assert!(!ledger.extension[&(owner, identity.browser.clone())].owner_parked); + assert!( + ledger + .begin_handoff_transfer(owner, &provider, &operation, epoch, "17") + .unwrap() + ); + ledger + .extension_context( + owner, + &provider, + operation.clone(), + epoch, + CallTarget::Tab { + origin: Origin::UserClaimed, + tab: Some("17".into()), + }, + true, + ) + .unwrap() + }; + let attach = async { + let frame = read_frame(&mut fixture.server).await; + assert_eq!(frame["request"]["method"], "attach"); + result( + &mut fixture.server, + frame["request"]["id"].as_u64().unwrap(), + json!({"claim":"replacement-claim","tabId":17,"targetId":"target-17"}), + ) + .await; + }; + let (attached, ()) = bounded(async { + tokio::join!( + replacement.dispatch(replacement.control.attach_tab(17)), + attach + ) + }) + .await; + attached.unwrap(); + { + let mut ledger = host.lock().await; + ledger.reconcile_provider(owner, &provider).await.unwrap(); + ledger + .apply_user_preparation( + owner, + &provider, + &identity.browser, + Some("17".into()), + PreparedOperation::Acquired { operation, epoch }, + serde_json::from_value(json!({"id":17})).unwrap(), + ) + .unwrap(); + assert!(matches!( + ledger.tabs[&identity].disposition, + Disposition::Live + )); + assert!(matches!( + ledger.tabs[&identity].origin, + Origin::AgentCreated + )); + assert_eq!( + ledger.tabs[&identity] + .extension + .as_ref() + .unwrap() + .claim + .as_ref() + .unwrap() + .claim_id(), + "replacement-claim" + ); + assert!(matches!( + ledger.tabs[&identity.browser.tab("18")].disposition, + Disposition::Live + )); + ledger.reconcile_provider(owner, &provider).await.unwrap(); + ledger.extension.clear(); + } + fixture.assert_no_frame_and_close().await; +} diff --git a/platform/computer-use/src/browser/host_handoff.rs b/platform/computer-use/src/browser/host_handoff.rs index 6f11af0e3a..1e6b2e0897 100644 --- a/platform/computer-use/src/browser/host_handoff.rs +++ b/platform/computer-use/src/browser/host_handoff.rs @@ -303,6 +303,14 @@ impl Browsers { tab.frame = None; tab.previous = None; } + super::extension::drain_extension_cleanup( + &host, + self.owner, + None, + Some(ExtensionAdapterMode::Chatgpt), + None, + ) + .await?; if let Some(turn) = turn { let finished = host .lock() diff --git a/platform/computer-use/src/browser/playwright/download_media.rs b/platform/computer-use/src/browser/playwright/download_media.rs index 351b5340de..d20a2c9316 100644 --- a/platform/computer-use/src/browser/playwright/download_media.rs +++ b/platform/computer-use/src/browser/playwright/download_media.rs @@ -33,7 +33,7 @@ impl Tab { ) -> Result { let cancellation = call.cancellation; let resources = Resources::new(); - let budget = super::super::downloads::locator_timeout(options); + let budget = super::super::downloads::timeout(options)?; let deadline = (!budget.is_zero()).then(|| tokio::time::Instant::now() + budget); let operation = async { loop { diff --git a/platform/computer-use/src/browser/playwright/evaluate.rs b/platform/computer-use/src/browser/playwright/evaluate.rs index 8a98aca206..b58a1a4ae4 100644 --- a/platform/computer-use/src/browser/playwright/evaluate.rs +++ b/platform/computer-use/src/browser/playwright/evaluate.rs @@ -203,6 +203,7 @@ impl Tab { let resources = Resources::new(); let bindings = Mutex::new(Vec::new()); let last_error = Mutex::new("playwright.evaluate exceeded its deadline".to_owned()); + let requested = options.budget()?; let timeout = || { let reason = last_error.lock().expect("evaluate error mutex poisoned"); if mode == BrowserEvaluateMode::Element @@ -210,7 +211,7 @@ impl Tab { { return failed(format!( "Timed out after {}ms evaluating selector {}: {}", - options.budget().as_secs_f64() * 1000.0, + requested.as_secs_f64() * 1000.0, locator.selector, reason.as_str() )); diff --git a/platform/computer-use/src/browser/playwright/mutation.rs b/platform/computer-use/src/browser/playwright/mutation.rs index ccf0a04ca0..d5e8bfb4c9 100644 --- a/platform/computer-use/src/browser/playwright/mutation.rs +++ b/platform/computer-use/src/browser/playwright/mutation.rs @@ -144,16 +144,8 @@ pub(in crate::browser) fn validate_mutation( Ok(()) } -fn timeout(operation: &BrowserMutationOperation) -> Duration { - let maximum = if matches!( - operation, - BrowserMutationOperation::PressSequentially { .. } - ) { - 5000.0 - } else { - 3000.0 - }; - Duration::from_secs_f64(operation.timeout_ms().unwrap_or(3000.0).min(maximum) / 1000.0) +fn timeout(operation: &BrowserMutationOperation) -> Result { + crate::protocol::timeout_duration(operation.timeout_ms().unwrap_or(3000.0)) } // Raw AX names have a small preprocessing layer before the shared key layout. @@ -578,7 +570,7 @@ impl Tab { let language = call.language; validate_mutation(&operation)?; let resources = Resources::new(); - let deadline = tokio::time::Instant::now() + timeout(&operation); + let deadline = tokio::time::Instant::now() + timeout(&operation)?; let context = MutationContext { document: &document, clock: MutationClock::Selector(deadline), diff --git a/platform/computer-use/src/browser/wait.rs b/platform/computer-use/src/browser/wait.rs index 50bea06411..37fe21d998 100644 --- a/platform/computer-use/src/browser/wait.rs +++ b/platform/computer-use/src/browser/wait.rs @@ -99,14 +99,8 @@ pub(super) fn js_string(value: &Value) -> String { _ => value.to_string(), } } -fn timeout(value: Option<&Value>) -> Duration { - Duration::from_secs_f64( - value - .and_then(Value::as_f64) - .unwrap_or(3000.0) - .clamp(0.0, 3000.0) - / 1000.0, - ) +fn timeout(value: Option<&Value>) -> Result { + crate::protocol::timeout_duration(value.and_then(Value::as_f64).unwrap_or(3000.0)) } struct Glob(regex::Regex); @@ -176,7 +170,7 @@ impl Plan { Self::Locator { locator, state, - timeout: timeout(timeout_ms.as_ref()), + timeout: timeout(timeout_ms.as_ref())?, } } BrowserWaitOperation::Url { @@ -199,12 +193,12 @@ impl Plan { matcher: Glob::new(&pattern)?, pattern, until, - timeout: timeout(timeout_ms.as_ref()), + timeout: timeout(timeout_ms.as_ref())?, } } BrowserWaitOperation::LoadState { state, timeout_ms } => Self::Load { state: LoadState::from_value(state.as_ref()), - timeout: timeout(timeout_ms.as_ref()), + timeout: timeout(timeout_ms.as_ref())?, }, BrowserWaitOperation::Timeout { timeout_ms } => { let value = timeout_ms @@ -213,13 +207,7 @@ impl Plan { .ok_or_else(|| { failed("playwright.waitForTimeout requires a non-negative integer") })?; - // The selected Node runtime's setTimeout treats zero and overflow as 1ms. - let millis = if !(1.0..=2_147_483_647.0).contains(&value) { - 1 - } else { - value as u64 - }; - Self::Delay(Duration::from_millis(millis)) + Self::Delay(crate::protocol::timeout_duration(value)?) } }) } diff --git a/platform/computer-use/src/desktop.rs b/platform/computer-use/src/desktop.rs index 4aa4939653..9803fe7ab8 100644 --- a/platform/computer-use/src/desktop.rs +++ b/platform/computer-use/src/desktop.rs @@ -274,12 +274,16 @@ impl SessionOwner { } /// Finalize shared resources after joining every original epoch close. - /// All observers retain the same original cleanup result. + /// Concurrent observers join the original cleanup; terminal failures may reconcile again. pub async fn close(&self) -> Result<(), ToolError> { let pending = { let mut state = self.state.lock().unwrap(); state.closed = true; - if let Some(pending) = &state.cleanup { + if let Some(pending) = state + .cleanup + .as_ref() + .filter(|pending| !matches!(*pending.borrow(), Some(Err(_)))) + { pending.clone() } else { let remote_live = !state.remote_epochs.is_empty(); @@ -850,6 +854,13 @@ impl Session { #[cfg(feature = "native-desktop")] Some(Backend::Local(local)) => local.close_native().await, Some(Backend::Windows { .. }) => { + if self + .remote_cleanup + .as_ref() + .is_some_and(|pending| matches!(*pending.borrow(), Some(Err(_)))) + { + self.remote_cleanup = None; + } let pending = self.remote_cleanup.get_or_insert_with(|| { let (reply, pending) = tokio::sync::watch::channel(None); let driver = self.driver.clone(); diff --git a/platform/computer-use/src/desktop/worker/epochs.rs b/platform/computer-use/src/desktop/worker/epochs.rs index 36b2340c4f..7084e061a2 100644 --- a/platform/computer-use/src/desktop/worker/epochs.rs +++ b/platform/computer-use/src/desktop/worker/epochs.rs @@ -49,7 +49,6 @@ type AudioObservations = FuturesUnordered)>>, identities: HashMap, - cleanup: Option, } impl PendingInvocations { fn insert( @@ -81,14 +80,6 @@ impl PendingInvocations { self.identities .remove(&id) .expect("parked completion retains its original request identity"); - if let Err( - error @ (ToolError::CleanupUnconfirmed(_) - | ToolError::OperationCleanupUnconfirmed { .. } - | ToolError::Persistence(_)), - ) = &result - { - self.cleanup.get_or_insert(error.clone()); - } let _ = events.send(Event::Response { id, result: result.map_err(Failure::from), @@ -108,7 +99,7 @@ impl PendingInvocations { { self.forward_next(events).await; } - self.cleanup.clone().map_or(Ok(()), Err) + Ok(()) } } fn epoch_cleanup( @@ -152,10 +143,14 @@ enum Epoch { Closed(NativeCleanup), } +type EpochTask = futures_util::future::Shared< + BoxFuture<'static, Result>, String>>, +>; + struct EpochWorker { session: String, queue: mpsc::UnboundedSender, - task: Option>, + task: EpochTask, closing: Option, } @@ -627,9 +622,12 @@ impl EpochWorker { pending.cancel(None); let pending_cleanup = pending.drain(None, &events).await; let native_cleanup = settle(local.close_native(), &disconnected).await; - let _ = reply - .send_replace(Some(epoch_cleanup(pending_cleanup, native_cleanup))); - break; + let result = epoch_cleanup(pending_cleanup, native_cleanup); + let closed = result.is_ok(); + let _ = reply.send_replace(Some(result)); + if closed { + break; + } } } } @@ -637,15 +635,30 @@ impl EpochWorker { let _ = pending.drain(None, &events).await; local }); + let task = async move { + task.await + .map(|local| Arc::new(Mutex::new(local))) + .map_err(|error| error.to_string()) + } + .boxed() + .shared(); + tokio::spawn(task.clone()); Self { session, queue, - task: Some(task), + task, closing: None, } } fn close(&mut self) -> NativeCleanup { + if self + .closing + .as_ref() + .is_some_and(|pending| matches!(*pending.borrow(), Some(Err(_)))) + { + self.closing = None; + } self.closing .get_or_insert_with(|| { let (reply, observation) = watch::channel(None); @@ -730,9 +743,9 @@ impl Epoch { }) } - fn take_task(&mut self) -> Option> { + fn task(&self) -> Option { match self { - Self::Open(worker) => worker.task.take(), + Self::Open(worker) => Some(worker.task.clone()), Self::Closed(_) => None, } } @@ -748,8 +761,7 @@ impl Epoch { let completed = match self { Self::Open(worker) => worker.closing.as_ref().and_then(|result| { let confirmed = matches!(result.borrow().as_ref(), Some(Ok(()))); - (confirmed && worker.task.as_ref().is_some_and(JoinHandle::is_finished)) - .then(|| result.clone()) + (confirmed && matches!(worker.task.peek(), Some(Ok(_)))).then(|| result.clone()) }), Self::Closed(_) => None, }; @@ -772,13 +784,19 @@ impl Owner { } fn close(&mut self) -> NativeCleanup { + if self + .closing + .as_ref() + .is_some_and(|pending| matches!(*pending.borrow(), Some(Err(_)))) + { + self.closing = None; + } self.closing .get_or_insert_with(|| { - let observations: Vec<_> = self.epochs.values_mut().map(Epoch::close).collect(); - let tasks: Vec<_> = self + let epochs: Vec<_> = self .epochs .values_mut() - .filter_map(Epoch::take_task) + .map(|epoch| (epoch.close(), epoch.task())) .collect(); let shared = std::mem::take(&mut self.shared); let resources = self.resources.clone(); @@ -793,19 +811,28 @@ impl Owner { )); } } - for observation in observations { - if let Err(error) = observe_retirement(observation).await { - failures.push(error.to_string()); - } - } let mut retained = Vec::new(); - for task in tasks { - match task.await { - Ok(local) => retained.push(local), - Err(error) => failures - .push(format!("Windows native epoch worker failed: {error}")), + for (observation, task) in epochs { + match observe_retirement(observation).await { + Err(error) => failures.push(error.to_string()), + Ok(()) => { + if let Some(task) = task { + match task.await { + Ok(local) => retained.push(local), + Err(error) => failures.push(format!( + "Windows native epoch worker failed: {error}" + )), + } + } + } } } + if !failures.is_empty() { + let _ = reply.send_replace(Some(Err(ToolError::CleanupUnconfirmed( + failures.join("; "), + )))); + return; + } if let Err(error) = settle(resources.close(), &disconnected).await { failures.push(error.to_string()); } @@ -1987,31 +2014,6 @@ mod tests { assert!(pending.is_empty()); } - #[tokio::test] - async fn a_forwarded_deferred_cleanup_failure_remains_in_epoch_close_custody() { - let (events, mut responses) = mpsc::unbounded_channel(); - let mut pending = PendingInvocations::default(); - pending.insert( - 13, - "turn".into(), - CancellationToken::new(), - Box::pin(async { - Err(ToolError::CleanupUnconfirmed( - "original Fetch disable reply was lost".into(), - )) - }), - ); - pending.forward_next(&events).await; - assert!( - matches!(responses.recv().await.unwrap(), Event::Response { id: 13, result: Err(Failure::CleanupUnconfirmed(message)), .. } if message == "original Fetch disable reply was lost") - ); - assert!(pending.is_empty()); - let result = epoch_cleanup(pending.drain(None, &events).await, Ok(())); - assert!( - matches!(result, Err(ToolError::CleanupUnconfirmed(message)) if message == "original Fetch disable reply was lost") - ); - } - fn preparation_identity() -> ScopeIdentity { ScopeIdentity { owner: "transport-owner".into(), diff --git a/platform/computer-use/src/driver.rs b/platform/computer-use/src/driver.rs index 44f26e898b..639748166a 100644 --- a/platform/computer-use/src/driver.rs +++ b/platform/computer-use/src/driver.rs @@ -20,9 +20,11 @@ pub(crate) mod native_computer; use crate::desktop::{NativeCleanup, NativeEpochId, observe_retirement}; +#[cfg(test)] +use cua_driver_sdk::TrustedSessionOptions; use cua_driver_sdk::{ - ConfiguredDriverOptions, CuaDriver, CuaDriverSession, DriverError, RuntimeAuthorizationOptions, - SessionPermissionMode, TrustedSessionOptions, worker::ActionCompletion, + ConfiguredDriverOptions, CuaDriver, CuaDriverSession, DriverError, HostSessionOptions, + RuntimeAuthorizationOptions, SessionPermissionMode, worker::ActionCompletion, }; use futures_util::future::BoxFuture; use maka_agent_facts::capability::CallResult; @@ -62,6 +64,8 @@ enum State { pub struct Driver { state: State, pub(crate) audio_enabled: bool, + #[cfg(all(target_os = "linux", not(target_env = "musl")))] + native_input: Arc>>, sessions: std::collections::HashMap>, #[cfg(target_os = "windows")] native_owner: crate::desktop::worker::native_owner::NativeOwner, @@ -581,7 +585,16 @@ impl Driver { ) -> BoxFuture<'static, Result<(), ToolError>> { self.reap_retired_epochs(); if let Some(retirement) = self.retiring_epochs.get(&epoch) { - return Box::pin(observe_retirement(retirement.pending.clone())); + let pending = retirement.pending.clone(); + #[cfg(all(target_os = "linux", not(target_env = "musl")))] + let failed = matches!(*pending.borrow(), Some(Err(_))); + #[cfg(all(target_os = "linux", not(target_env = "musl")))] + let pending = if failed { + self.retry_failed_raw_retirement(epoch) + } else { + pending + }; + return Box::pin(observe_retirement(pending)); } let bound = self.sessions.remove(&epoch); let session = bound.as_ref().map(|bound| bound.session.clone()); @@ -651,8 +664,8 @@ impl Driver { Box::pin(observe_retirement(pending)) } - // Terminal Host cleanup only. A new attempt releases the original fixed - // resources; it cannot resume the old user action or change its receipt. + // Retry only the original resources after a terminal cleanup failure. + // The old user action and its receipt are never replayed. #[cfg(all(target_os = "linux", not(target_env = "musl")))] fn retry_failed_raw_retirement(&mut self, epoch: NativeEpochId) -> NativeCleanup { let retirement = self @@ -1127,15 +1140,12 @@ impl Driver { std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(), std::collections::hash_map::Entry::Vacant(entry) => { let session = driver - .create_trusted_session(TrustedSessionOptions { + .create_host_session(HostSessionOptions { // SDK lifecycle tombstones belong to their original transport. // A reset owns a fresh episode, never revives an old lease's ID. public_session: format!("maka-{}", uuid::Uuid::new_v4()), mode: SessionPermissionMode::Standard, - ttl_seconds: 86400, - idle_ttl_seconds: 86400, capability_manifest_path: None, - bounded_manifest_path: None, }) .map_err(|error| native_error(error, language))?; entry.insert(Arc::new(BoundSession { @@ -1202,6 +1212,8 @@ impl Driver { } self.sessions.clear(); self.retiring_epochs.clear(); + #[cfg(all(target_os = "linux", not(target_env = "musl")))] + self.native_input.lock().await.take(); #[cfg(target_os = "macos")] self.mac_controllers.clear(); self.state = State::Closed; @@ -1594,9 +1606,9 @@ mod tests { .await .unwrap(); assert_ne!(before.session, other.session); - assert!( - before.expires_in_seconds > 300, - "bound lifecycle must use its explicit idle lease" + assert_eq!( + before.expires_in_seconds, None, + "bound lifecycle must follow its host owner" ); driver.release_epoch(a_epoch).await.unwrap(); assert!( diff --git a/platform/computer-use/src/driver/native_computer.rs b/platform/computer-use/src/driver/native_computer.rs index 6ccd7e91ad..879dbe530a 100644 --- a/platform/computer-use/src/driver/native_computer.rs +++ b/platform/computer-use/src/driver/native_computer.rs @@ -28,6 +28,13 @@ pub(super) struct NativeTurns { closed: std::collections::HashSet, } +#[cfg(all(target_os = "linux", not(target_env = "musl")))] +pub(super) struct ActiveInput { + bound: Arc, + turn: String, + owner: Arc, +} + impl Driver { /// These are actual registered Cua tools. This narrow entry adds no model /// named-tool escape hatch to the ordinary App facade. @@ -77,6 +84,28 @@ impl Driver { } let bound = driver.lock().await.bind_epoch(epoch, language)?; let session = bound.ready(language).await?; + let device = driver.lock().await.native_input.clone(); + let mut device = device.lock().await; + if let Some(previous) = device.as_ref() + && (!Arc::ptr_eq(&previous.bound, &bound) || previous.turn != turn) + { + previous.owner.seal(); + previous.owner.close().await.map_err(|error| { + ToolError::Failed(format!( + "native_input_busy: previous input release is unconfirmed: {error}" + )) + })?; + let mut turns = previous.bound.native_turns.lock().await; + if turns + .owners + .get(&previous.turn) + .is_some_and(|owner| Arc::ptr_eq(owner, &previous.owner)) + { + turns.owners.remove(&previous.turn); + } + drop(turns); + *device = None; + } let owner = { let mut turns = bound.native_turns.lock().await; if turns.closed.contains(turn) { @@ -95,6 +124,11 @@ impl Driver { .clone(), } }; + *device = Some(ActiveInput { + bound: bound.clone(), + turn: turn.to_owned(), + owner: owner.clone(), + }); if cancellation.is_cancelled() { return Err(ToolError::Failed( "Computer Use cancelled before native input dispatch".into(), @@ -313,8 +347,8 @@ mod retirement_tests { .is_ok(), "joining a failed resource cleanup must not close the original SDK session" ); - // A separate terminal close is the sole retry admission. It reuses - // that exact passive owner, then drains/closes the real SDK binding. + driver.release_epoch(epoch).await.unwrap(); + assert!(bound.native_turns.lock().await.owners.is_empty()); driver.close().await.unwrap(); assert!(matches!(driver.state, State::Closed)); assert!(bound.native_turns.lock().await.owners.is_empty()); diff --git a/platform/computer-use/src/protocol.rs b/platform/computer-use/src/protocol.rs index 80976a34a5..ad2f298d14 100644 --- a/platform/computer-use/src/protocol.rs +++ b/platform/computer-use/src/protocol.rs @@ -19,6 +19,23 @@ use serde::{Deserialize, Serialize}; +#[cfg(feature = "native-desktop")] +pub(crate) fn timeout_duration( + milliseconds: f64, +) -> Result { + let invalid = || { + maka_agent_facts::tools::ToolError::Failed( + "timeoutMs is outside the supported duration range".into(), + ) + }; + let duration = + std::time::Duration::try_from_secs_f64(milliseconds / 1000.0).map_err(|_| invalid())?; + tokio::time::Instant::now() + .checked_add(duration) + .ok_or_else(invalid)?; + Ok(duration) +} + pub(crate) mod browser_capability; pub use browser_capability::{ BundleOptions as PageAssetBundleOptions, Viewport as BrowserViewport, @@ -399,8 +416,7 @@ pub enum BrowserEvaluateMode { All, } -/// Evaluate preserves the selected runtime's numeric timeout clamp independently -/// of locator read option validation. Other JavaScript values use the default. +/// JavaScript evaluation's mechanical I/O budget. #[derive(Debug, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct BrowserEvaluateOptions { @@ -408,14 +424,13 @@ pub struct BrowserEvaluateOptions { } impl BrowserEvaluateOptions { #[cfg(feature = "native-desktop")] - pub(crate) fn budget(&self) -> std::time::Duration { + pub(crate) fn budget(&self) -> Result { let milliseconds = self .timeout_ms .as_ref() .and_then(serde_json::Value::as_f64) - .unwrap_or(3000.0) - .clamp(0.0, 3000.0); - std::time::Duration::from_secs_f64(milliseconds / 1000.0) + .unwrap_or(3000.0); + timeout_duration(milliseconds) } } diff --git a/platform/computer-use/src/protocol/browser_filechooser.rs b/platform/computer-use/src/protocol/browser_filechooser.rs index 0937029e21..d5f5b15efa 100644 --- a/platform/computer-use/src/protocol/browser_filechooser.rs +++ b/platform/computer-use/src/protocol/browser_filechooser.rs @@ -22,8 +22,7 @@ use serde_json::Value; #[cfg(feature = "native-desktop")] use std::time::Duration; -/// Chooser schema validation precedes browser dispatch. Explicit null is not a -/// missing timeout; very large finite integral JS numbers still clamp to 3s. +/// Chooser schema validation precedes browser dispatch. Explicit null is not a missing timeout. #[derive(Clone, Debug, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct BrowserFileChooserOptions { @@ -51,6 +50,6 @@ impl BrowserFileChooserOptions { ) })?, }; - Ok(Duration::from_secs_f64(millis.min(3000.0) / 1000.0)) + super::timeout_duration(millis) } } diff --git a/platform/computer-use/tests/integration/repl_playwright/mod.rs b/platform/computer-use/tests/integration/repl_playwright/mod.rs index 51eaa033b1..3d4ef67102 100644 --- a/platform/computer-use/tests/integration/repl_playwright/mod.rs +++ b/platform/computer-use/tests/integration/repl_playwright/mod.rs @@ -68,8 +68,13 @@ impl BrowserAuthorization for SiteAuthority { if cancellation.is_cancelled() { return Err(ToolError::Failed("fixture cancelled".into())); } - if !allowed || access.origin()?.as_deref() != Some(expected.as_str()) { - return Err(ToolError::Failed("fixture site denied".into())); + if !allowed + || (!matches!(access, BrowserAccess::Metadata { .. }) + && access.origin()?.as_deref() != Some(expected.as_str())) + { + return Err(ToolError::Failed(format!( + "fixture site denied: expected {expected}, access {access:?}" + ))); } Ok(()) }) diff --git a/platform/computer-use/tests/integration/repl_playwright/readonly_evaluate.rs b/platform/computer-use/tests/integration/repl_playwright/readonly_evaluate.rs index cc1090d7f6..8668cf1202 100644 --- a/platform/computer-use/tests/integration/repl_playwright/readonly_evaluate.rs +++ b/platform/computer-use/tests/integration/repl_playwright/readonly_evaluate.rs @@ -88,6 +88,11 @@ async fn public_readonly_evaluate_keeps_scope_redaction_and_never_replays_user_c 7, "a failed user evaluation must not automatically execute again" ); + let delayed = evaluate(&repl, bridge.clone(), r#" + const delayed=await evalPage.evaluate(()=>new Promise(resolve=>setTimeout(()=>resolve('ready after the former limit'),3200)),undefined,{timeoutMs:8000}); + nodeRepl.write({delayed}); + "#).await; + assert_eq!(delayed["delayed"], "ready after the former limit"); authority.allowed.store(false, Ordering::SeqCst); let denied=evaluate(&repl,bridge.clone(),r#" let denied=false;try{await evalPage.evaluate(()=>document.title);}catch(error){denied=error.message.includes('fixture site denied');} diff --git a/plugins/agent/native/src/owner/tests/service_tools.rs b/plugins/agent/native/src/owner/tests/service_tools.rs index 94363f1cd6..fe7a2c6faf 100644 --- a/plugins/agent/native/src/owner/tests/service_tools.rs +++ b/plugins/agent/native/src/owner/tests/service_tools.rs @@ -56,6 +56,7 @@ fn counter_contract() -> Contract { #[derive(Clone, Copy)] enum Behavior { Returned, + ComputerClosed, Unknown, CleanupFailed, Blocked, @@ -85,6 +86,9 @@ impl Handler for Counter { input: Value, context: Context, ) -> BoxFuture<'static, Result> { + if matches!(self.behavior, Behavior::ComputerClosed) { + return Box::pin(async { Err("computer_closed".into()) }); + } let count = if method == "read" { assert!(context.operation().is_none()); self.effects.load(Ordering::SeqCst) @@ -102,6 +106,9 @@ impl Handler for Counter { entered.add_permits(1); match behavior { Behavior::Returned => Ok(json!({"effects":count})), + Behavior::ComputerClosed => { + unreachable!("unavailable Computer returns before effects") + } Behavior::Unknown => Err("lost reply after original effect".into()), Behavior::CleanupFailed => { let mut ticket = context.scope().resources.reserve().unwrap(); @@ -195,6 +202,45 @@ pub(super) async fn pending_approval( } } +#[tokio::test] +async fn ordinary_conversation_survives_an_unavailable_computer_service() { + tokio::time::timeout(Duration::from_secs(30), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}/v1", listener.local_addr().unwrap()); + let fixture = Fixture::with_services( + &base, + None, + None, + Some(Setup::computer(Counter::new(Behavior::ComputerClosed))), + ) + .await; + let session = fixture.create("Continue without browser").await; + fixture.send(&session).await.unwrap(); + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + assert!( + sent["tools"] + .as_array() + .unwrap() + .iter() + .any(|tool| tool["function"]["name"] == "Counter") + ); + response_with_text(&mut socket, "We can continue discussing the task.").await; + fixture.drained().await; + let facts = facts(&fixture, &session).await; + assert!(facts.iter().any(|fact| matches!( + fact, + Fact::InvocationEnded { + outcome: maka_agent_facts::event::InvocationOutcome::Completed + } + ))); + fixture.app.shutdown().await.unwrap(); + fixture.config.shutdown().await.unwrap(); + }) + .await + .unwrap(); +} + #[tokio::test] async fn live_tool_input_replays_across_model_completion_approval_and_original_effect_settlement() { tokio::time::timeout(Duration::from_secs(30), async { diff --git a/plugins/agent/native/src/owner/tests/service_tools/setup.rs b/plugins/agent/native/src/owner/tests/service_tools/setup.rs index 9c0814f5c5..79fa8e7ef7 100644 --- a/plugins/agent/native/src/owner/tests/service_tools/setup.rs +++ b/plugins/agent/native/src/owner/tests/service_tools/setup.rs @@ -24,6 +24,7 @@ enum Service { handler: Option>, query: bool, }, + Computer(Arc), Questions, Forms, File(&'static str), @@ -164,6 +165,9 @@ impl Setup { pub(in crate::owner::tests) fn jev() -> Self { Self::new(Service::Jev) } + pub(super) fn computer(handler: Arc) -> Self { + Self::new(Service::Computer(handler)) + } pub(in crate::owner::tests) fn web() -> Self { Self::new(Service::Web) } @@ -289,7 +293,9 @@ impl Setup { }, title: None, confirmation: self.confirmation, - result: if matches!(self.service, Service::File("readImage")) + result: if matches!(self.service, Service::Computer(_)) { + contract::ToolResult::Computer + } else if matches!(self.service, Service::File("readImage")) || matches!(&self.service, Service::Images(_)) { contract::ToolResult::Image @@ -461,6 +467,7 @@ impl Service { fn contract(&self) -> Contract { match self { Self::Counter { .. } => counter_contract(), + Self::Computer(_) => maka_workspace_api::computer::contract(), Self::Questions => maka_interactions_api::questions::contract(), Self::Forms => maka_interactions_api::forms::contract(), Self::File(_) | Self::FileEditors => maka_workspace_api::files::contract(), @@ -476,6 +483,7 @@ impl Service { fn export(&self) -> &str { match self { Self::Counter { .. } => "Counter", + Self::Computer(_) => maka_workspace_api::computer::EXPORT, Self::Questions => maka_interactions_api::questions::EXPORT, Self::Forms => maka_interactions_api::forms::EXPORT, Self::File(_) | Self::FileEditors => maka_workspace_api::files::EXPORT, @@ -490,6 +498,7 @@ impl Service { } fn method(&self) -> &str { match self { + Self::Computer(_) => "evaluate", Self::Counter { query: true, .. } => "read", Self::Counter { .. } => "write", Self::Questions | Self::Forms => "ask", @@ -510,6 +519,7 @@ impl Service { return json!({"kind":"javascript","module":"counter.mjs","realm":"shared"}); } Self::Counter { .. } => "fixture.counter", + Self::Computer(_) => maka_workspace_api::computer::IMPLEMENTATION, Self::Questions => maka_interactions_api::questions::IMPLEMENTATION, Self::Forms => maka_interactions_api::forms::IMPLEMENTATION, Self::File(_) | Self::FileEditors => maka_workspace_api::files::IMPLEMENTATION, @@ -560,6 +570,7 @@ impl Service { .. }) => &["prepare", "search"], Self::Web => &["capture"], + Self::Computer(_) => &["state"], #[cfg(target_os = "macos")] Self::Protocol => &["run", "input"], _ => &[], @@ -594,6 +605,29 @@ impl Service { .unwrap(); } Self::Counter { handler: None, .. } => {} + Self::Computer(handler) => { + let handler = handler.clone(); + registry + .insert( + maka_workspace_api::computer::IMPLEMENTATION.into(), + Arc::new(move |_, _| { + let handler = handler.clone(); + Box::pin(async move { + NativeActivation { + exports: Ok(BTreeMap::from([( + maka_workspace_api::computer::EXPORT.into(), + Published { + contract: maka_workspace_api::computer::contract(), + value: handler.clone() as Arc, + }, + )])), + owner: handler, + } + }) + }), + ) + .unwrap(); + } Self::Questions => maka_interactions_native::questions::provide(registry).unwrap(), Self::Forms => maka_interactions_native::forms::provide(registry).unwrap(), Self::File(_) | Self::FileEditors => maka_files_native::provide(registry).unwrap(), diff --git a/plugins/agent/native/src/tools.rs b/plugins/agent/native/src/tools.rs index f3c4aa6877..a5e29bfff2 100644 --- a/plugins/agent/native/src/tools.rs +++ b/plugins/agent/native/src/tools.rs @@ -146,7 +146,6 @@ pub(crate) fn catalog( let questions_contract = maka_interactions_api::questions::contract(); let web_contract = maka_web_api::contract(); let mut web_bindings = std::collections::BTreeMap::>::new(); - let mut computer_bindings = std::collections::BTreeMap::>::new(); let mut shell_bindings = Vec::new(); let mut skill_bindings = Vec::new(); let skill_contract = maka_skills_api::contract(); @@ -310,12 +309,6 @@ pub(crate) fn catalog( if selection.input == ToolInput::Skill { skill_bindings.push((selection.name.clone(), Arc::new(SkillBinding(service.clone())))); } - if selection.result == ToolResult::Computer { - computer_bindings - .entry(selection.target.clone()) - .or_default() - .push(service.clone()); - } if selection.input == ToolInput::Shell { shell_bindings.push(( selection.name.clone(), @@ -385,21 +378,6 @@ pub(crate) fn catalog( ) .map_err(|_| "agent_web_binding_invalid")?; } - for services in computer_bindings.into_values() { - let names = services - .iter() - .map(|service| service.selection.name.clone()) - .collect(); - catalog = catalog - .with_request_binding( - names, - Arc::new(computer::Binding { - services, - workers: workers.clone(), - }), - ) - .map_err(|_| "agent_computer_binding_invalid")?; - } for (name, binding) in shell_bindings { catalog = catalog .with_request_binding([name].into(), binding) diff --git a/plugins/agent/native/src/tools/computer.rs b/plugins/agent/native/src/tools/computer.rs index 35f66b2611..b9e08b8530 100644 --- a/plugins/agent/native/src/tools/computer.rs +++ b/plugins/agent/native/src/tools/computer.rs @@ -25,9 +25,7 @@ use maka_agent_facts::{ tool_output::ToolContent, }; use maka_code_runtime::CellOutput; -use maka_tools::request::{BindingRequest, BoundTool, RequestBinding}; -use maka_workspace_api::computer::{ReplState, ResultValue}; -use std::collections::BTreeMap; +use maka_workspace_api::computer::ResultValue; pub(super) fn title(method: &str, input: &Value) -> maka_agent_facts::display::Text { use maka_agent_facts::display::Text; @@ -44,99 +42,6 @@ pub(super) fn title(method: &str, input: &Value) -> maka_agent_facts::display::T Text::localized("Use computer", "操作电脑", "操作電腦") } -pub(super) struct Binding { - pub services: Vec, - pub workers: tokio_util::task::TaskTracker, -} -impl RequestBinding for Binding { - fn bind( - &self, - input: BindingRequest, - ) -> futures_util::future::BoxFuture<'static, Result, ToolError>> - { - let services: Vec<_> = self - .services - .iter() - .filter(|service| input.tools.contains(&service.selection.name)) - .cloned() - .collect(); - let service = services - .first() - .expect("selected request binding group") - .clone(); - let workers = self.workers.clone(); - Box::pin(async move { - let request = service - .services - .request_with_execution( - Identity::Agent { - invocation: input.invocation, - operation_id: None, - }, - input.cancellation.clone(), - input.execution.ok_or_else(|| { - ToolError::Persistence("Execution custody unavailable".into()) - })?, - ) - .map_err(ToolError::Failed)?; - let target = service.selection.target.clone(); - let state = workers - .spawn(async move { - let result = AssertUnwindSafe(async { - request - .call_target( - &target, - "state".into(), - serde_json::json!({}), - Attachments::default(), - )? - .await - }) - .catch_unwind() - .await; - request - .finish() - .await - .map_err(ToolError::CleanupUnconfirmed)?; - result - .map_err(|_| ToolError::Failed("Computer state capture panicked".into()))? - .map_err(ToolError::Failed) - }) - .await - .map_err(|_| { - ToolError::CleanupUnconfirmed("Computer state worker disappeared".into()) - })??; - let state: ReplState = serde_json::from_value(state) - .map_err(|_| ToolError::Failed("Computer state invalid".into()))?; - let hint = match state { - ReplState::Fresh => { - "Current REPL state: fresh. Earlier transcript variables and target bindings are unavailable." - } - ReplState::Ready => { - "Current REPL state: ready. Existing variables and target bindings are retained in this Host process." - } - ReplState::ResetRequired => { - "Current REPL state: reset_required. Call cua_reset and rebind targets. Dispatched input may already have taken effect." - } - }; - Ok(services - .into_iter() - .map(|service| { - let name = service.selection.name.clone(); - let description = format!("{}\n{}", service.selection.description, hint); - ( - name, - BoundTool::Described { - handler: Arc::new(service), - description, - }, - ) - }) - .collect()) - }) - } -} - pub(super) async fn project( store: &SessionStore, invocation: &Invocation, @@ -189,7 +94,6 @@ async fn parts( if record.source != ArtifactSource::SessionEffect || record.kind != ArtifactKind::File || record.mime_type.as_deref() != Some("application/json") - || record.size_bytes > 50 * 1024 * 1024 { return Err(ToolError::Failed( "Computer output does not belong to the original invocation".into(), diff --git a/plugins/workspace/workspace-computer/native/src/admission.rs b/plugins/workspace/workspace-computer/native/src/admission.rs index 8e173abe94..90c8928d65 100644 --- a/plugins/workspace/workspace-computer/native/src/admission.rs +++ b/plugins/workspace/workspace-computer/native/src/admission.rs @@ -222,18 +222,23 @@ pub(super) struct Operation { lease: Option, ticket: Option, tasks: tokio_util::task::TaskTracker, - stopping: tokio_util::sync::CancellationToken, + gate: Arc, } impl Operation { pub fn new( owner: &Arc, lease: InvocationEffectLease, native: &Arc, + gate: &Arc, ) -> Result { if owner.stopping.is_cancelled() { lease.finish(Ok(())); // No journal dispatch or native effect was admitted. return Err(ToolError::Failed("Computer Host service is closed".into())); } + if let Err(error) = gate.check_open() { + lease.finish(Ok(())); + return Err(error); + } let mut ticket = match owner.resources.reserve() { Ok(ticket) => ticket, Err(error) => { @@ -252,7 +257,7 @@ impl Operation { lease: Some(lease), ticket: Some(ticket), tasks: owner.tasks.clone(), - stopping: owner.stopping.clone(), + gate: gate.clone(), }) } @@ -281,7 +286,7 @@ impl Operation { root: self.root.take(), lease: Some(lease), ticket: self.ticket.take(), - stopping: self.stopping.clone(), + gate: self.gate.clone(), failure: None, observations: Vec::new(), }; @@ -311,7 +316,7 @@ struct PendingSettlement { root: Option>, lease: Option, ticket: Option, - stopping: tokio_util::sync::CancellationToken, + gate: Arc, failure: Option, observations: Vec, } @@ -319,7 +324,7 @@ impl PendingSettlement { fn fail(&mut self, observation: ToolError) { let failure = observation.to_string(); self.observations.push(observation); - self.stopping.cancel(); + self.gate.close_admission(); if let Some(lease) = self.lease.take() { lease.retain_unconfirmed(failure.clone()); } @@ -483,6 +488,7 @@ impl BrowserEffectScope for AttachedEffectScope { } struct HostAdmission { + gate: Arc, owner: Arc, native: Arc, receipt: Receipt, @@ -510,6 +516,7 @@ impl HostEffects { }; ticket.start(); Ok(HostAdmission { + gate: self.gate.clone(), owner, native, receipt, @@ -525,6 +532,7 @@ impl HostEffects { preparation_active: Option, ) { let HostAdmission { + gate, owner, native, receipt, @@ -537,7 +545,7 @@ impl HostEffects { failure: None, observations: Vec::new(), ticket: Some(ticket), - stopping: owner.stopping.clone(), + gate, root: Some(owner.store.clone()), completed, preparation_active, @@ -722,7 +730,7 @@ struct HostCustody { failure: Option, observations: Vec, ticket: Option, - stopping: tokio_util::sync::CancellationToken, + gate: Arc, root: Option>, completed: Option>>, preparation_active: Option, @@ -754,7 +762,7 @@ impl HostCustody { } let error = observation.to_string(); self.observations.push(observation); - self.stopping.cancel(); + self.gate.close_admission(); if let Some(ticket) = &self.ticket { ticket.record_failure(error.clone()); } @@ -828,7 +836,7 @@ impl Drop for HostCustody { active.cancel(); } if let Some(pending) = self.pending.take() { - self.stopping.cancel(); + self.gate.close_admission(); if let Some(ticket) = self.ticket.take() { ticket.unconfirmed( "Computer Host-service effect owner disappeared".into(), @@ -1003,7 +1011,13 @@ mod tests { #[tokio::test] async fn durable_closed_result_retains_same_run_until_original_rpc_is_ready() { let fixture = Fixture::new().await; - let operation = Operation::new(&fixture.owner, fixture.lease(), &fixture.native).unwrap(); + let operation = Operation::new( + &fixture.owner, + fixture.lease(), + &fixture.native, + &Arc::default(), + ) + .unwrap(); let handoff = operation.handoff(); let release = CancellationToken::new(); let _release_on_failure = release.clone().drop_guard(); @@ -1073,7 +1087,8 @@ mod tests { #[tokio::test] async fn real_uncertainty_signal_fences_admission_but_drains_the_same_host_future() { let fixture = Fixture::new().await; - let host = host_effects(&fixture.owner, Arc::default(), &fixture.native); + let gate = Arc::new(Gate::default()); + let host = host_effects(&fixture.owner, gate.clone(), &fixture.native); let started = CancellationToken::new(); let uncertain = CancellationToken::new(); let release = CancellationToken::new(); @@ -1104,12 +1119,14 @@ mod tests { .await .expect("admitted Host future was not polled"); uncertain.cancel(); - tokio::time::timeout( - std::time::Duration::from_secs(5), - fixture.owner.stopping.cancelled(), - ) - .await - .expect("actual uncertainty was not recorded"); + tokio::time::timeout(std::time::Duration::from_secs(5), gate.wait_closed()) + .await + .expect("actual uncertainty was not recorded"); + assert!(!fixture.owner.stopping.is_cancelled()); + let sibling = host_effects(&fixture.owner, Arc::default(), &fixture.native); + sibling + .spawn(Box::pin(async { Ok(()) })) + .expect("another VM remains usable"); let rejected_calls = calls.clone(); assert!( host.spawn(Box::pin(async move { @@ -1122,7 +1139,7 @@ mod tests { assert_eq!(calls.load(AtomicOrdering::SeqCst), 1); assert!( matches!( - Operation::new(&fixture.owner, fixture.lease(), &fixture.native), + Operation::new(&fixture.owner, fixture.lease(), &fixture.native, &gate), Err(ToolError::Failed(_)) ), "uncertain real owner must refuse a fresh operation before native or journal dispatch" @@ -1154,7 +1171,9 @@ mod tests { #[tokio::test] async fn checked_rpc_uncertainty_quarantines_the_original_invocation_before_other_rpc_drains() { let fixture = Fixture::new().await; - let operation = Operation::new(&fixture.owner, fixture.lease(), &fixture.native).unwrap(); + let gate = Arc::new(Gate::default()); + let operation = + Operation::new(&fixture.owner, fixture.lease(), &fixture.native, &gate).unwrap(); let handoff = operation.handoff(); let fail_first = CancellationToken::new(); let release_last = CancellationToken::new(); @@ -1200,12 +1219,10 @@ mod tests { .unwrap(); operation.complete(None); fail_first.cancel(); - tokio::time::timeout( - std::time::Duration::from_secs(5), - fixture.owner.stopping.cancelled(), - ) - .await - .expect("actual uncertainty was not recorded"); + tokio::time::timeout(std::time::Duration::from_secs(5), gate.wait_closed()) + .await + .expect("actual uncertainty was not recorded"); + assert!(!fixture.owner.stopping.is_cancelled()); assert!( fixture.effects.settle().await.is_err(), "the real original invocation must retain unconfirmed custody" @@ -1228,7 +1245,13 @@ mod tests { #[tokio::test] async fn closed_checked_handoff_returns_the_original_unpolled_future() { let fixture = Fixture::new().await; - let operation = Operation::new(&fixture.owner, fixture.lease(), &fixture.native).unwrap(); + let operation = Operation::new( + &fixture.owner, + fixture.lease(), + &fixture.native, + &Arc::default(), + ) + .unwrap(); let handoff = operation.handoff(); operation.complete(None); let calls = Arc::new(AtomicUsize::new(0)); diff --git a/plugins/workspace/workspace-computer/native/src/evaluate.rs b/plugins/workspace/workspace-computer/native/src/evaluate.rs index d162401114..63f4442b7b 100644 --- a/plugins/workspace/workspace-computer/native/src/evaluate.rs +++ b/plugins/workspace/workspace-computer/native/src/evaluate.rs @@ -490,7 +490,8 @@ impl Computer { let effect_owner = self.clone(); let effect_store = lease.store(); let journal_store = effect_store.clone(); - let settlement = admission::Operation::new(self, lease, &bridge.epoch.native)?; + let settlement = + admission::Operation::new(self, lease, &bridge.epoch.native, &bridge.admission)?; let operation_settlement = settlement.handoff(); let host_effects = admission::host_effects_with_lease_children( self, @@ -651,7 +652,7 @@ impl Computer { // its canonical fence remains a Host persistence/cleanup // failure; a physical drain must not wash that failure. receipt.fail(error.clone()); - self.stopping.cancel(); + bridge.admission.close_admission(); settlement.complete(Some(error.clone())); return Err(error.into()); } @@ -671,7 +672,7 @@ impl Computer { )) = &result { receipt.fail(error.clone()); - self.stopping.cancel(); + bridge.admission.close_admission(); Some(error.clone()) } else { None @@ -858,7 +859,7 @@ impl ToolExecutor for Bridge { | ToolError::OperationCleanupUnconfirmed { .. } ) { receipt.fail(error.clone()); - bridge.owner.stopping.cancel(); + bridge.admission.close_admission(); } else { receipt.complete_call(); } @@ -1071,16 +1072,17 @@ mod tests { } _ => panic!("unexpected bridge failure for {error:?}: {result:?}"), } - assert!(owner.stopping.is_cancelled()); + assert!(!owner.stopping.is_cancelled()); assert!(admission.check_open().is_err()); } runtime.close().await.unwrap(); drop(bridge); + kernel.close(CancellationToken::new()).await.unwrap(); owner.sessions.close().await.unwrap(); owner.resources.settle().await.unwrap(); owner.tasks.close(); owner.tasks.wait().await; - drop((kernel, interaction, owner)); + drop((interaction, owner)); Arc::try_unwrap(root).ok().unwrap().close().await.unwrap(); } } diff --git a/plugins/workspace/workspace-computer/native/src/lib.rs b/plugins/workspace/workspace-computer/native/src/lib.rs index 1a0e09bda3..e2d957a545 100644 --- a/plugins/workspace/workspace-computer/native/src/lib.rs +++ b/plugins/workspace/workspace-computer/native/src/lib.rs @@ -213,7 +213,6 @@ impl Handler for Service { result, Err(ToolError::Persistence(_) | ToolError::CleanupUnconfirmed(_)) ) { - owner.stopping.cancel(); Err("Computer cleanup unconfirmed".into()) } else { Ok(()) @@ -230,8 +229,7 @@ impl Handler for Service { impl NativeOwner for Service { fn execution_basis(&self) -> Option { // Grants and output artifacts belong to this Root; device selection is - // captured configuration. REPL bindings are transient and their current - // state is captured separately for each model request. + // captured configuration. REPL state belongs to actual tool calls. let root = self.0.store.owned_root().ok()?; serde_json::to_value((root.root_id(), &self.0.configuration)).ok() } diff --git a/plugins/workspace/workspace-computer/native/src/sessions.rs b/plugins/workspace/workspace-computer/native/src/sessions.rs index 9fd25931d9..0d149afda5 100644 --- a/plugins/workspace/workspace-computer/native/src/sessions.rs +++ b/plugins/workspace/workspace-computer/native/src/sessions.rs @@ -78,7 +78,10 @@ impl TurnCleanup { work: impl std::future::Future> + Send + 'static, ) -> Retirement { let mut pending = self.pending.lock().unwrap(); - if let Some(pending) = pending.as_ref() { + if let Some(pending) = pending + .as_ref() + .filter(|pending| !matches!(pending.peek(), Some(Err(_)))) + { return pending.clone(); } let (completed, result) = tokio::sync::oneshot::channel(); @@ -310,6 +313,7 @@ pub(super) struct Epoch { turns: Arc>>>, pub(super) effects_changed: Arc, retirement: Mutex>, + vm_joined: std::sync::atomic::AtomicBool, } struct BrowserPreparationReceipt { basis: PermissionBasis, @@ -370,11 +374,6 @@ impl Sessions { Ok(existing.clone()) }; } - if entries.len() >= 16 { - return Err(failed( - "Computer Use session capacity reached; retire an unused Session", - )); - } let interaction = Arc::new(Interaction { kernel: Default::default(), native_owner: maka_computer_use::SessionOwner::new(self.driver.clone()), @@ -453,7 +452,6 @@ pub(super) async fn watch_turn( let session = invocation.session_id.clone(); let turn = invocation.turn_id.clone(); let stopping = owner.stopping.clone(); - let weak = Arc::downgrade(owner); owner.tasks.spawn(async move { let observed = async { loop { @@ -480,12 +478,6 @@ pub(super) async fn watch_turn( .finish_turn(&turn_cleanup, &session, &turn, language) .await; let result = observed.and(cleanup); - if result.is_err() { - entry.retiring.cancel(); - if let Some(owner) = weak.upgrade() { - owner.stopping.cancel(); - } - } entry.release_turn(&turn, &turn_cleanup, &result); ticket.complete(result.map_err(message)); }); @@ -517,10 +509,6 @@ pub(super) async fn drain_ended_turns( if turn_ended(&store, session, &turn).await? { let result = entry.finish_turn(&cleanup, session, &turn, language).await; entry.release_turn(&turn, &cleanup, &result); - if result.is_err() { - entry.retiring.cancel(); - owner.stopping.cancel(); - } result?; } } @@ -710,6 +698,9 @@ impl Kernel { { failures.push(format!("JavaScript runtime cleanup failed: {error}")); } + self.epoch + .vm_joined + .store(true, std::sync::atomic::Ordering::Release); if let Err(error) = self.epoch.settle_audio(&closing).await { failures.push(format!("Computer audio cleanup failed: {error}")); } @@ -745,6 +736,7 @@ impl Interaction { turns: self.turns.clone(), effects_changed: self.effects_changed.clone(), retirement: Default::default(), + vm_joined: std::sync::atomic::AtomicBool::new(false), }); epochs.push(epoch.clone()); Ok(Kernel { @@ -784,7 +776,6 @@ impl Interaction { kernel.admission.close_admission(); ticket.start(); let entry = Arc::downgrade(self); - let weak = Arc::downgrade(owner); let retired = epoch.clone(); let task = owner.tasks.spawn(async move { let result = AssertUnwindSafe(kernel.close(closing)) @@ -803,13 +794,6 @@ impl Interaction { .unwrap() .retain(|current| !Arc::ptr_eq(current, &retired)); } - } else { - if let Some(entry) = entry.upgrade() { - entry.retiring.cancel(); - } - if let Some(owner) = weak.upgrade() { - owner.stopping.cancel(); - } } ticket.complete(result.clone().map_err(message)); result @@ -829,7 +813,10 @@ impl Interaction { self.retiring.cancel(); let closing = { let mut closed = self.closed.lock().unwrap(); - if let Some(closing) = closed.as_ref() { + if let Some(closing) = closed + .as_ref() + .filter(|closing| !matches!(closing.peek(), Some(Err(_)))) + { closing.clone() } else { let entry = self.clone(); @@ -937,7 +924,10 @@ impl Interaction { .observation .lock() .expect("Browser cleanup observation lock poisoned"); - if observation.is_some() { + if observation + .as_ref() + .is_some_and(|original| !matches!(original.peek(), Some(Err(_)))) + { return Ok(()); } let mut ticket = owner.resources.reserve().map_err(failed)?; @@ -968,27 +958,72 @@ impl Interaction { "Browser preparation cleanup observer panicked".into(), )) }); - if result.is_err() { - entry.retiring.cancel(); - owner.stopping.cancel(); - } ticket.complete(result.clone().map_err(message)); result }); - *observation = Some( - async move { - task.await.map_err(|error| { - ToolError::CleanupUnconfirmed(format!( - "Browser cleanup observer disappeared: {error}", - )) - })? - } - .boxed() - .shared(), - ); + let result = async move { + task.await.map_err(|error| { + ToolError::CleanupUnconfirmed(format!( + "Browser cleanup observer disappeared: {error}" + )) + })? + } + .boxed() + .shared(); + tasks.spawn(result.clone()); + *observation = Some(result); Ok(()) } + async fn settle_epoch(self: &Arc, epoch: &Arc) -> Result<(), ToolError> { + let retirement = { + let mut pending = epoch + .retirement + .lock() + .expect("Computer epoch retirement lock poisoned"); + if let Some(original) = pending + .as_ref() + .filter(|original| !matches!(original.peek(), Some(Err(_)))) + { + original.clone() + } else { + if !epoch.vm_joined.load(std::sync::atomic::Ordering::Acquire) { + return Err(ToolError::CleanupUnconfirmed( + "Computer VM termination remains unconfirmed".into(), + )); + } + let epoch = epoch.clone(); + let entry = Arc::downgrade(self); + let task = self.tasks.spawn(async move { + let result = epoch.close_native().await; + if result.is_ok() { + epoch.clear_audio(); + if let Some(entry) = entry.upgrade() { + entry + .epochs + .lock() + .unwrap() + .retain(|current| !Arc::ptr_eq(current, &epoch)); + } + } + result + }); + let result = async move { + task.await.map_err(|error| { + ToolError::CleanupUnconfirmed(format!( + "Computer epoch cleanup worker disappeared: {error}" + )) + })? + } + .boxed() + .shared(); + *pending = Some(result.clone()); + result + } + }; + retirement.await + } + async fn retire_prepared_epoch( self: &Arc, owner: &Arc, @@ -999,8 +1034,8 @@ impl Interaction { .lock() .expect("Computer epoch retirement lock poisoned") .clone(); - if let Some(retirement) = retirement { - return retirement.await; + if retirement.is_some() { + return self.settle_epoch(epoch).await; } if self.retiring.is_cancelled() { return self.close().await; @@ -1011,9 +1046,9 @@ impl Interaction { .lock() .expect("Computer epoch retirement lock poisoned") .clone(); - if let Some(retirement) = retirement { + if retirement.is_some() { drop(current); - return retirement.await; + return self.settle_epoch(epoch).await; } if self.retiring.is_cancelled() { drop(current); @@ -1039,7 +1074,7 @@ impl Interaction { retirement.await } - async fn close_owned(&self) -> Result<(), ToolError> { + async fn close_owned(self: &Arc) -> Result<(), ToolError> { self.retiring.cancel(); let current = self.kernel.lock().await.take(); let mut failures = Vec::new(); @@ -1058,22 +1093,8 @@ impl Interaction { // the caller. Only these exact epoch workers belong to this close. let epochs = self.epochs.lock().unwrap().clone(); for epoch in epochs { - let retirement = epoch.retirement.lock().unwrap().clone(); - if let Some(retirement) = retirement { - if let Err(error) = retirement.await { - failures.push(message(error)); - } - } else { - match epoch.close_native().await { - Ok(()) => { - epoch.clear_audio(); - self.epochs - .lock() - .unwrap() - .retain(|current| !Arc::ptr_eq(current, &epoch)); - } - Err(error) => failures.push(message(error)), - } + if let Err(error) = self.settle_epoch(&epoch).await { + failures.push(message(error)); } } if let Err(error) = self.native_owner.close().await { @@ -1184,21 +1205,19 @@ fn watch_scanned(owner: &Arc, scanned: impl Fn() + Send + 'static) { ) } .await; - if !matches!(valid, Ok(true)) { + if let Err(error) = &valid { + eprintln!("Computer catalog observation failed for {id}: {error}"); + } + if matches!(valid, Ok(false)) { let Ok(mut ticket) = owner.resources.reserve() else { return; }; ticket.start(); let result = owner.sessions.retire(&id, &entry).await.map_err(message); - let failed = result.is_err(); - if failed { - entry.retiring.cancel(); - owner.stopping.cancel(); + if let Err(error) = &result { + eprintln!("Computer Session cleanup failed for {id}: {error}"); } ticket.complete(result); - if failed { - return; - } continue; } let epochs = entry.epochs.lock().unwrap().clone(); @@ -1214,13 +1233,12 @@ fn watch_scanned(owner: &Arc, scanned: impl Fn() + Send + 'static) { if matches!(observed, Ok(PreparationAuthority::Current)) { continue; } - if entry - .observe_browser_cleanup(&owner, &id, &epoch, &receipt, observed) - .is_err() + if let Err(error) = + entry.observe_browser_cleanup(&owner, &id, &epoch, &receipt, observed) { - entry.retiring.cancel(); - owner.stopping.cancel(); - return; + eprintln!( + "Computer browser cleanup observation failed for {id}: {error}" + ); } } let receipt = epoch.audio.lock().unwrap().clone(); @@ -1240,15 +1258,10 @@ fn watch_scanned(owner: &Arc, scanned: impl Fn() + Send + 'static) { Ok(true) => observed.map(|_| ()).map_err(message), Err(error) => Err(message(error)), }; - let failed = result.is_err(); - if failed { - entry.retiring.cancel(); - owner.stopping.cancel(); + if let Err(error) = &result { + eprintln!("Computer audio cleanup failed for {id}: {error}"); } ticket.complete(result); - if failed { - return; - } } } drop(owner); @@ -1317,6 +1330,36 @@ mod tests { assert!(current.audio.lock().unwrap().is_none()); } + #[tokio::test] + async fn cleanup_retries_after_failure_and_joins_an_inflight_retry() { + let tasks = tokio_util::task::TaskTracker::new(); + let cleanup = TurnCleanup::default(); + assert!( + cleanup + .start(&tasks, async { + Err(ToolError::CleanupUnconfirmed( + "temporary transport loss".into(), + )) + }) + .await + .is_err() + ); + let (release, released) = tokio::sync::oneshot::channel(); + let retry = cleanup.start(&tasks, async move { + released.await.expect("original cleanup release"); + Ok(()) + }); + let joined = cleanup.start(&tasks, async { + panic!("pending cleanup must be joined rather than replaced"); + }); + assert!(futures_util::poll!(Box::pin(retry.clone())).is_pending()); + release.send(()).unwrap(); + retry.await.unwrap(); + joined.await.unwrap(); + tasks.close(); + tasks.wait().await; + } + struct PreparedBrowserFixture { _directory: tempfile::TempDir, root: Arc,