From 4fd2d3d3be717c3f3ca4afbb93beb30a0b8db395 Mon Sep 17 00:00:00 2001 From: zhom <2717306+zhom@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:07:23 +0400 Subject: [PATCH] fix: remote dialogs and rest --- AGENTS.md | 3 +- e2e/tests/browser.test.mjs | 49 ++ package.json | 1 + sdk/README.md | 2 +- sdk/api-paths.json | 17 +- sdk/node/src/client.mts | 54 ++ sdk/node/src/coverage.mts | 3 + sdk/node/src/types.mts | 50 +- sdk/node/test/requests.test.mts | 24 + sdk/python/src/donutbrowser/client.py | 57 ++ sdk/python/src/donutbrowser/coverage.py | 3 + sdk/python/src/donutbrowser/models.py | 69 +- sdk/python/tests/test_requests.py | 30 + src-tauri/src/api_server.rs | 169 ++++- src-tauri/src/cdp_target.rs | 93 ++- src-tauri/src/mcp_server.rs | 241 +++++- src-tauri/src/mcp_server_fleet_tests.rs | 459 +++++++++++ src-tauri/src/page_dialog.rs | 967 +++++++++++++++++++++--- 18 files changed, 2118 insertions(+), 173 deletions(-) create mode 100644 src-tauri/src/mcp_server_fleet_tests.rs diff --git a/AGENTS.md b/AGENTS.md index 5b673313..59c65d42 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -65,7 +65,7 @@ donutbrowser/ │ │ ├── mcp_toolsets.rs # Optional MCP toolsets: which tools a client lists (core + opt-in sets) │ │ ├── mcp_views.rs # Compact MCP views: profile summaries, filters, paging, size caps │ │ ├── navigation_watch.rs # Whether a CDP action navigated, so a click waits only for a load that comes -│ │ ├── page_dialog.rs # Native dialogs on agent-driven pages: one Page-enabled socket per page, DIALOG_OPEN fail fast, answers +│ │ ├── page_dialog.rs # Native dialogs on agent-driven pages: a Page-enabled socket per local page, one relay socket per cloud session that pins the page its tools drive, DIALOG_OPEN fail fast, answers │ │ ├── network_capture.rs # MCP network capture: one browser-level socket per profile, in memory, credentials redacted │ │ ├── agent_uploads.rs # Files an agent sends to file_upload, kept until the browser stops │ │ ├── mcp_remote.rs # Remote MCP bridge: outbound websocket to Donut cloud (Enterprise remote control) @@ -174,6 +174,7 @@ suite passes: | Sync client/server, manifests, timestamps, deletion, encryption, password rollover | `pnpm e2e:sync` | | Wayfern download/terms/fingerprint, browser runner, CDP, automation endpoints, process cleanup | `pnpm e2e:browser` | | `donut-sync/` server code (routes, auth, S3 client, presigning, subscriptions) | `cd donut-sync && cargo test` against the compose S3 server | +| Cloud (remote) sessions: relay targets, the page a session's tools drive, dialogs and typing there | `pnpm test:fleet` with `DONUT_FLEET_TEST_KEY` (a `dsk_` key of a service account); it starts a real cloud session, so it is not in `pnpm e2e` | | E2E harness, WebDriver plugin/driver, app isolation hooks, or changes spanning multiple rows | Run every affected row; use `pnpm e2e` for cross-cutting changes | `e2e:browser` requires `WAYFERN_TEST_TOKEN` in the environment or local `.env`. `e2e:network` diff --git a/e2e/tests/browser.test.mjs b/e2e/tests/browser.test.mjs index 7c52671d..6393662e 100644 --- a/e2e/tests/browser.test.mjs +++ b/e2e/tests/browser.test.mjs @@ -1957,6 +1957,55 @@ test("MCP answers native dialogs and types into a background tab", async (t) => await tool("handle_dialog", { accept: true, prompt_text: "Bob" }); assert.equal(await cdp.evaluate("document.body.dataset.renamed"), "Bob"); + // The same over REST: reading is never refused, every other agent call is + // refused at once, and the answer reports what it answered. + const agentUrl = (action) => + `${base}/v1/profiles/${profile.id}/agent/${action}`; + await tool("click_locator", { + locator: { role: "button", name: "Delete" }, + }); + const restRead = await request(agentUrl("dialog"), { + token: saved.api_token, + }); + assert.equal(restRead.response.status, 200, JSON.stringify(restRead.value)); + assert.equal(restRead.value.open, true); + assert.equal(restRead.value.dialog.type, "confirm"); + const restRefused = await timed(() => + request(agentUrl("type"), { + method: "POST", + token: saved.api_token, + body: { locator: { role: "textbox", name: "Later" }, text: "x" }, + }), + ); + assert.equal( + restRefused.value.response.status, + 409, + JSON.stringify(restRefused.value.value), + ); + assert.equal(restRefused.value.value.code, "DIALOG_OPEN"); + assert.equal(restRefused.value.value.dialog.message, "Delete it?"); + assert.ok(restRefused.ms < 5_000, `REST type waited ${restRefused.ms} ms`); + const restAnswered = await request(agentUrl("handle-dialog"), { + method: "POST", + token: saved.api_token, + body: { accept: true }, + }); + assert.equal( + restAnswered.response.status, + 200, + JSON.stringify(restAnswered.value), + ); + assert.equal(restAnswered.value.accepted, true); + assert.equal(restAnswered.value.dialog.message, "Delete it?"); + assert.equal(await cdp.evaluate("document.body.dataset.deleted"), "true"); + const restNone = await request(agentUrl("handle-dialog"), { + method: "POST", + token: saved.api_token, + body: { accept: true }, + }); + assert.equal(restNone.response.status, 409); + assert.equal(restNone.value.code, "NO_DIALOG_OPEN"); + // The person brings another tab to the front, and the agent's tab loads // again behind it. The text still reaches the field, and the answer says // how. diff --git a/package.json b/package.json index de3ca684..7922fa74 100644 --- a/package.json +++ b/package.json @@ -43,6 +43,7 @@ "test:rust": "cd src-tauri && cargo test", "test:donut-sync": "cd donut-sync && cargo test --lib --locked", "test:rust:unit": "cd src-tauri && cargo test --lib && cargo test --test donut_proxy_integration && cargo test --test vpn_integration", + "test:fleet": "cd src-tauri && cargo test --lib fleet_ -- --ignored --nocapture", "test:sync-e2e": "node scripts/sync-test-harness.mjs", "e2e": "node e2e/run.mjs --suite=full", "e2e:smoke": "node e2e/run.mjs --suite=smoke", diff --git a/sdk/README.md b/sdk/README.md index 0e8f5234..eb2ed9f5 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -223,7 +223,7 @@ server-side failure. | 403 | `Forbidden` | `Forbidden` | Wayfern terms not accepted, or not signed in | | 404 | `NotFound` | `NotFound` | No entity with that id | | 408 | `RequestTimeout` | `RequestTimeout` | `agent/pick` waited and nothing was picked | -| 409 | `Conflict` | `Conflict` | A browser, a teammate or a remote session holds the profile | +| 409 | `Conflict` | `Conflict` | A browser, a teammate or a remote session holds the profile; a native dialog holds the page (`DIALOG_OPEN`: read it with `agent_dialog` / `agentDialog`, answer it with `agent_handle_dialog` / `agentHandleDialog`) | | 429 | `RateLimited` | `RateLimited` | Automation quota spent; `retry_after` / `retryAfter` | | 500 | `ServerError` | `ServerError` | Internal failure | | 502 | `BadGateway` | `BadGateway` | The browser or the relay answered wrongly | diff --git a/sdk/api-paths.json b/sdk/api-paths.json index 928606aa..d634f559 100644 --- a/sdk/api-paths.json +++ b/sdk/api-paths.json @@ -2,7 +2,7 @@ "source": "src-tauri/src/api_server.rs", "regenerate_with": "python3 sdk/tools/extract-api-paths.py", "description": "Every operation the desktop app publishes in its /openapi.json. The SDK test suites assert this list and their own coverage tables match exactly, so an endpoint added to the app fails the SDK tests until it is either wrapped or deliberately listed as omitted.", - "operation_count": 85, + "operation_count": 88, "operations": [ { "operation_id": "download_browser_api", @@ -214,11 +214,26 @@ "method": "POST", "path": "/v1/profiles/{id}/agent/click" }, + { + "operation_id": "agent_dialog_api", + "method": "GET", + "path": "/v1/profiles/{id}/agent/dialog" + }, { "operation_id": "agent_extract_api", "method": "POST", "path": "/v1/profiles/{id}/agent/extract" }, + { + "operation_id": "agent_handle_dialog_api", + "method": "POST", + "path": "/v1/profiles/{id}/agent/handle-dialog" + }, + { + "operation_id": "agent_hold_api", + "method": "POST", + "path": "/v1/profiles/{id}/agent/hold" + }, { "operation_id": "agent_perceive_api", "method": "POST", diff --git a/sdk/node/src/client.mts b/sdk/node/src/client.mts index b9bbda38..b4857418 100644 --- a/sdk/node/src/client.mts +++ b/sdk/node/src/client.mts @@ -11,6 +11,8 @@ import { DonutConnectionError, DonutError, errorForStatus } from "./errors.mts"; import type { AgentClick, + AgentDialogAnswer, + AgentHold, AgentTyping, ApiBulkCreateResponse, ApiGroupResponse, @@ -47,6 +49,7 @@ import type { ImportProxiesResponse, LocatorDescription, LocatorResolution, + PageDialog, PerceptionPage, PickedElement, ProfileImportBatchResult, @@ -607,6 +610,29 @@ export class DonutClient { ); } + /** + * POST /v1/profiles/{id}/agent/hold + * + * Name the element with `locator` or `selector`. `duration_ms` is 100 to + * 30000, default 3000; `button` is `"left"` (the default), `"middle"` or + * `"right"`. + */ + agentHold( + profileId: string, + request: { + locator?: LocatorDescription; + selector?: string; + duration_ms?: number; + button?: string; + }, + ): Promise { + return this.#request( + "POST", + `/v1/profiles/${segment(profileId)}/agent/hold`, + body({ ...request }), + ); + } + /** * POST /v1/profiles/{id}/agent/type * @@ -672,6 +698,34 @@ export class DonutClient { ); } + /** + * GET /v1/profiles/{id}/agent/dialog + * + * Whether a native dialog holds the page, and which. Never refused for the + * dialog: while one is open, every other agent call throws `DIALOG_OPEN`. + */ + agentDialog(profileId: string): Promise { + return this.#request("GET", `/v1/profiles/${segment(profileId)}/agent/dialog`); + } + + /** + * POST /v1/profiles/{id}/agent/handle-dialog + * + * `accept` true is OK, false is Cancel. An accepted prompt takes + * `prompt_text`, or keeps its default text. Throws `NO_DIALOG_OPEN` when + * nothing is open. + */ + agentHandleDialog( + profileId: string, + request: { accept: boolean; prompt_text?: string }, + ): Promise { + return this.#request( + "POST", + `/v1/profiles/${segment(profileId)}/agent/handle-dialog`, + body({ ...request }), + ); + } + // ------------------------------------------------------------------ // Remote sessions // ------------------------------------------------------------------ diff --git a/sdk/node/src/coverage.mts b/sdk/node/src/coverage.mts index 7d7fb138..f26c868f 100644 --- a/sdk/node/src/coverage.mts +++ b/sdk/node/src/coverage.mts @@ -62,7 +62,10 @@ export const OPERATIONS: ReadonlyMap = new Map([ ["GET /v1/profiles/{id}/fingerprint", "getProfileFingerprint"], ["PUT /v1/profiles/{id}", "updateProfile"], ["POST /v1/profiles/{id}/agent/click", "agentClick"], + ["GET /v1/profiles/{id}/agent/dialog", "agentDialog"], ["POST /v1/profiles/{id}/agent/extract", "agentExtract"], + ["POST /v1/profiles/{id}/agent/handle-dialog", "agentHandleDialog"], + ["POST /v1/profiles/{id}/agent/hold", "agentHold"], ["POST /v1/profiles/{id}/agent/perceive", "agentPerceive"], ["POST /v1/profiles/{id}/agent/pick", "agentPick"], ["POST /v1/profiles/{id}/agent/resolve-locator", "agentResolveLocator"], diff --git a/sdk/node/src/types.mts b/sdk/node/src/types.mts index 20431f7a..8e049f00 100644 --- a/sdk/node/src/types.mts +++ b/sdk/node/src/types.mts @@ -8,8 +8,8 @@ * snake_case; the browser-facing agent types (`LocatorDescription`, * `LocatorCandidate`, `PerceptionPage` and friends) carry the browser's own * camelCase, because they are handed through from the browser rather than - * restated. `AgentClick` and `AgentTyping` are the exceptions inside the agent - * surface: they are snake_case with a single `match` key. These types follow + * restated. `AgentClick`, `AgentHold` and `AgentTyping` are the exceptions + * inside the agent surface: they are snake_case with a single `match` key. These types follow * the wire rather than tidying it, so a value read from one call can be passed * straight into the next. */ @@ -800,3 +800,49 @@ export interface AgentTyping { */ typed_with: "browser_keyboard" | "key_events" | "inserted"; } + +/** What a press and hold did. Note the snake_case body and the `match` key. */ +export interface AgentHold { + held: boolean; + /** The element a locator settled on. Absent when a selector named it. */ + match?: LocatorCandidate; + engine: Engine; + /** Whether a page load followed the release. */ + navigated: boolean; + button: string; + /** How long the button was down, in milliseconds. */ + held_ms: number; +} + +/** A native dialog (alert, confirm, prompt, "leave this page?") open on a page. */ +export interface OpenDialog { + /** `alert`, `confirm`, `prompt` or `beforeunload`. */ + type: string; + message: string; + /** The text a prompt puts in its field. */ + default_prompt: string; + /** The page that opened it. */ + url: string; +} + +/** Whether a native dialog holds the page. */ +export interface PageDialog { + /** + * True while a dialog holds the page, false when none does, null when the + * page answers nothing and no dialog was seen opening on it. + */ + open: boolean | null; + dialog?: OpenDialog; + /** False when the page answers nothing. A navigation ends that. */ + page_responding?: boolean; + note?: string; +} + +/** What answering a dialog did. */ +export interface AgentDialogAnswer { + accepted: boolean; + /** The dialog that was answered. */ + dialog: OpenDialog; + /** A dialog the page opened in answer. It holds the page now. */ + next?: OpenDialog; +} diff --git a/sdk/node/test/requests.test.mts b/sdk/node/test/requests.test.mts index ba974ea5..b7804a70 100644 --- a/sdk/node/test/requests.test.mts +++ b/sdk/node/test/requests.test.mts @@ -349,6 +349,30 @@ const CASES: Case[] = [ body: { timeout_ms: 15000 }, operation: "POST /v1/profiles/{id}/agent/pick", }, + { + method: "agentHold", + args: ["p1", { selector: "#press", duration_ms: 4000 }], + verb: "POST", + path: "/v1/profiles/p1/agent/hold", + body: { selector: "#press", duration_ms: 4000 }, + operation: "POST /v1/profiles/{id}/agent/hold", + }, + { + method: "agentDialog", + args: ["p1"], + verb: "GET", + path: "/v1/profiles/p1/agent/dialog", + body: null, + operation: "GET /v1/profiles/{id}/agent/dialog", + }, + { + method: "agentHandleDialog", + args: ["p1", { accept: true, prompt_text: "Bob" }], + verb: "POST", + path: "/v1/profiles/p1/agent/handle-dialog", + body: { accept: true, prompt_text: "Bob" }, + operation: "POST /v1/profiles/{id}/agent/handle-dialog", + }, // -- remote sessions ----------------------------------------------------- { method: "listRemoteSessions", diff --git a/sdk/python/src/donutbrowser/client.py b/sdk/python/src/donutbrowser/client.py index 6f42fb79..1486c742 100644 --- a/sdk/python/src/donutbrowser/client.py +++ b/sdk/python/src/donutbrowser/client.py @@ -21,6 +21,8 @@ from urllib.parse import quote, urlencode, urlsplit from .errors import DonutConnectionError, DonutError, error_for_status from .models import ( AgentClick, + AgentDialogAnswer, + AgentHold, AgentTyping, ApiBulkCreateResponse, ApiGroupResponse, @@ -57,6 +59,7 @@ from .models import ( ImportProxiesResponse, LocatorDescription, LocatorResolution, + PageDialog, PerceptionPage, PickedElement, ProfileImportBatchResult, @@ -666,6 +669,32 @@ class DonutClient: body=_body(locator=dict(locator), button=button, click_count=click_count), ) + def agent_hold( + self, + profile_id: str, + *, + locator: Optional[LocatorDescription] = None, + selector: Optional[str] = None, + duration_ms: Optional[float] = None, + button: Optional[str] = None, + ) -> AgentHold: + """POST /v1/profiles/{id}/agent/hold + + Name the element with ``locator`` or ``selector``. ``duration_ms`` is + 100 to 30000, default 3000; ``button`` is ``"left"`` (the default), + ``"middle"`` or ``"right"``. + """ + return self._request( + "POST", + f"/v1/profiles/{_segment(profile_id)}/agent/hold", + body=_body( + locator=dict(locator) if locator is not None else None, + selector=selector, + duration_ms=duration_ms, + button=button, + ), + ) + def agent_type( self, profile_id: str, @@ -739,6 +768,34 @@ class DonutClient: body=_body(timeout_ms=timeout_ms), ) + def agent_dialog(self, profile_id: str) -> PageDialog: + """GET /v1/profiles/{id}/agent/dialog + + Whether a native dialog holds the page, and which. Never refused for + the dialog: while one is open, every other agent call raises + ``DIALOG_OPEN``. + """ + return self._request("GET", f"/v1/profiles/{_segment(profile_id)}/agent/dialog") + + def agent_handle_dialog( + self, + profile_id: str, + *, + accept: bool, + prompt_text: Optional[str] = None, + ) -> AgentDialogAnswer: + """POST /v1/profiles/{id}/agent/handle-dialog + + ``accept`` true is OK, false is Cancel. An accepted prompt takes + ``prompt_text``, or keeps its default text. Raises ``NO_DIALOG_OPEN`` + when nothing is open. + """ + return self._request( + "POST", + f"/v1/profiles/{_segment(profile_id)}/agent/handle-dialog", + body=_body(accept=accept, prompt_text=prompt_text), + ) + # ------------------------------------------------------------------ # Remote sessions # ------------------------------------------------------------------ diff --git a/sdk/python/src/donutbrowser/coverage.py b/sdk/python/src/donutbrowser/coverage.py index c0b25c97..7aa48fbf 100644 --- a/sdk/python/src/donutbrowser/coverage.py +++ b/sdk/python/src/donutbrowser/coverage.py @@ -69,7 +69,10 @@ OPERATIONS: Dict[Operation, str] = { ("GET", "/v1/profiles/{id}/fingerprint"): "get_profile_fingerprint", ("PUT", "/v1/profiles/{id}"): "update_profile", ("POST", "/v1/profiles/{id}/agent/click"): "agent_click", + ("GET", "/v1/profiles/{id}/agent/dialog"): "agent_dialog", ("POST", "/v1/profiles/{id}/agent/extract"): "agent_extract", + ("POST", "/v1/profiles/{id}/agent/handle-dialog"): "agent_handle_dialog", + ("POST", "/v1/profiles/{id}/agent/hold"): "agent_hold", ("POST", "/v1/profiles/{id}/agent/perceive"): "agent_perceive", ("POST", "/v1/profiles/{id}/agent/pick"): "agent_pick", ("POST", "/v1/profiles/{id}/agent/resolve-locator"): "agent_resolve_locator", diff --git a/sdk/python/src/donutbrowser/models.py b/sdk/python/src/donutbrowser/models.py index 476ce7ee..2b799a3f 100644 --- a/sdk/python/src/donutbrowser/models.py +++ b/sdk/python/src/donutbrowser/models.py @@ -9,8 +9,8 @@ Two spellings live side by side because the app sends both. Most bodies are snake_case; the browser-facing agent types (``LocatorDescription``, ``LocatorCandidate``, ``PerceptionPage`` and friends) carry the browser's own camelCase, because they are handed through from the browser rather than -restated. ``AgentClick`` and ``AgentTyping`` are the exceptions inside the -agent surface: they are snake_case with a single ``match`` key. The types below +restated. ``AgentClick``, ``AgentHold`` and ``AgentTyping`` are the exceptions +inside the agent surface: they are snake_case with a single ``match`` key. The types below follow the wire rather than tidying it, so a value read from one call can be passed straight into the next. """ @@ -94,7 +94,11 @@ __all__ = [ "Extraction", "PickedElement", "AgentClick", + "AgentHold", "AgentTyping", + "OpenDialog", + "PageDialog", + "AgentDialogAnswer", ] # The app's own JSON for a proxy's settings. Declared `Object` in the OpenAPI @@ -931,3 +935,64 @@ class AgentTyping(_AgentTypingRequired, total=False): """ corrections: int + + +class _AgentHoldRequired(TypedDict): + held: bool + engine: str + navigated: bool + button: str + held_ms: int + + +class AgentHold(_AgentHoldRequired, total=False): + """What a press and hold did. Note the snake_case body and the ``match`` key. + + ``match`` is the element a locator settled on, absent when a selector named + it. ``held_ms`` is how long the button was down. + """ + + match: LocatorCandidate + + +class OpenDialog(TypedDict): + """A native dialog (alert, confirm, prompt, "leave this page?") on a page. + + ``type`` is ``alert``, ``confirm``, ``prompt`` or ``beforeunload``; + ``default_prompt`` is the text a prompt puts in its field; ``url`` is the + page that opened it. + """ + + type: str + message: str + default_prompt: str + url: str + + +class _PageDialogRequired(TypedDict): + open: Optional[bool] + + +class PageDialog(_PageDialogRequired, total=False): + """Whether a native dialog holds the page. + + ``open`` is true while a dialog holds the page, false when none does, and + ``None`` when the page answers nothing and no dialog was seen opening on + it; ``page_responding`` is then false. A navigation ends that. + """ + + dialog: OpenDialog + page_responding: bool + note: str + + +class _AgentDialogAnswerRequired(TypedDict): + accepted: bool + dialog: OpenDialog + + +class AgentDialogAnswer(_AgentDialogAnswerRequired, total=False): + """What answering a dialog did. ``next`` is a dialog the page opened in + answer, which holds the page now.""" + + next: OpenDialog diff --git a/sdk/python/tests/test_requests.py b/sdk/python/tests/test_requests.py index b2394927..8916fd6e 100644 --- a/sdk/python/tests/test_requests.py +++ b/sdk/python/tests/test_requests.py @@ -365,6 +365,36 @@ CASES: List[Case] = [ {}, "/v1/profiles/{id}/agent/pick", ), + ( + "agent_hold", + ("p1",), + {"selector": "#press", "duration_ms": 4000.0}, + "POST", + "/v1/profiles/p1/agent/hold", + {"selector": "#press", "duration_ms": 4000.0}, + {}, + "/v1/profiles/{id}/agent/hold", + ), + ( + "agent_dialog", + ("p1",), + {}, + "GET", + "/v1/profiles/p1/agent/dialog", + None, + {}, + "/v1/profiles/{id}/agent/dialog", + ), + ( + "agent_handle_dialog", + ("p1",), + {"accept": True, "prompt_text": "Bob"}, + "POST", + "/v1/profiles/p1/agent/handle-dialog", + {"accept": True, "prompt_text": "Bob"}, + {}, + "/v1/profiles/{id}/agent/handle-dialog", + ), # -- remote sessions --------------------------------------------------- ( "list_remote_sessions", diff --git a/src-tauri/src/api_server.rs b/src-tauri/src/api_server.rs index 2cba2ddb..ec579ed9 100644 --- a/src-tauri/src/api_server.rs +++ b/src-tauri/src/api_server.rs @@ -885,6 +885,8 @@ struct ImportProxiesResponse { agent_type_api, agent_extract_api, agent_pick_api, + agent_dialog_api, + agent_handle_dialog_api, ), components(schemas( ApiProfile, @@ -1005,6 +1007,10 @@ struct ImportProxiesResponse { crate::mcp_server::AgentTyping, crate::mcp_server::TypedWith, crate::mcp_server::AgentPickRequest, + crate::mcp_server::PageDialog, + crate::mcp_server::AgentDialogRequest, + crate::mcp_server::AgentDialogAnswer, + crate::page_dialog::OpenDialog, )), tags( (name = "profiles", description = "Profile management endpoints"), @@ -1265,6 +1271,8 @@ fn build_v1_router() -> Router { .routes(routes!(agent_type_api)) .routes(routes!(agent_extract_api)) .routes(routes!(agent_pick_api)) + .routes(routes!(agent_dialog_api)) + .routes(routes!(agent_handle_dialog_api)) .split_for_parts(); // The two paths that carry an extension payload, kept apart so the raised @@ -1455,7 +1463,16 @@ fn is_automation_request(method: &Method, path: &str) -> bool { // exactly as a script one is. Some(_), Some("agent"), - Some("perceive" | "resolve-locator" | "click" | "hold" | "type" | "extract" | "pick"), + Some( + "perceive" + | "resolve-locator" + | "click" + | "hold" + | "type" + | "extract" + | "pick" + | "handle-dialog" + ), None ) ) @@ -5464,9 +5481,21 @@ async fn check_browser_downloaded( // resolved the way every automation endpoint is: 402 without the plan, 404 for // an unknown profile, 409 when it is not running, 429 from the shared limiter. -/// The running browser behind `id`, and the engine its version entitles it to. +/// The running browser behind `id`, and the engine its version entitles it to, +/// refused while a dialog holds its page. async fn agent_context_for( id: &str, +) -> Result { + let ctx = agent_target_for(id).await?; + crate::mcp_server::page_ready(&ctx.target, false) + .await + .map_err(agent_error_response)?; + Ok(ctx) +} + +/// [`agent_context_for`], whatever the page is doing. +async fn agent_target_for( + id: &str, ) -> Result { if !crate::cloud_auth::CLOUD_AUTH .can_use_browser_automation() @@ -5585,7 +5614,7 @@ fn agent_error_response(error: crate::mcp_server::AgentError) -> (StatusCode, St (status = 401, description = "Unauthorized"), (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), - (status = 409, description = "The profile is not running"), + (status = 409, description = "The profile is not running, or a native dialog holds the page (`code` DIALOG_OPEN; read it with GET agent/dialog, answer it with agent/handle-dialog) or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5616,7 +5645,7 @@ async fn agent_perceive_api( (status = 401, description = "Unauthorized"), (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), - (status = 409, description = "The profile is not running"), + (status = 409, description = "The profile is not running, or a native dialog holds the page (`code` DIALOG_OPEN; read it with GET agent/dialog, answer it with agent/handle-dialog) or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5651,7 +5680,7 @@ async fn agent_resolve_locator_api( (status = 401, description = "Unauthorized"), (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), - (status = 409, description = "The profile is not running"), + (status = 409, description = "The profile is not running, or a native dialog holds the page (`code` DIALOG_OPEN; read it with GET agent/dialog, answer it with agent/handle-dialog) or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5686,7 +5715,7 @@ async fn agent_click_api( (status = 401, description = "Unauthorized"), (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), - (status = 409, description = "The profile is not running"), + (status = 409, description = "The profile is not running, or a native dialog holds the page (`code` DIALOG_OPEN; read it with GET agent/dialog, answer it with agent/handle-dialog) or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5717,7 +5746,7 @@ async fn agent_hold_api( (status = 401, description = "Unauthorized"), (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), - (status = 409, description = "The profile is not running, the field did not take the focus or the text (`code` TYPE_NOT_APPLIED, `reason` not_focused, focus_moved or value), or a dialog is open on the page (`code` DIALOG_OPEN)"), + (status = 409, description = "The profile is not running, the field did not take the focus or the text (`code` TYPE_NOT_APPLIED, `reason` not_focused, focus_moved or value), a native dialog holds the page (`code` DIALOG_OPEN), or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5753,7 +5782,7 @@ async fn agent_type_api( (status = 401, description = "Unauthorized"), (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), - (status = 409, description = "The profile is not running"), + (status = 409, description = "The profile is not running, or a native dialog holds the page (`code` DIALOG_OPEN; read it with GET agent/dialog, answer it with agent/handle-dialog) or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5785,7 +5814,7 @@ async fn agent_extract_api( (status = 402, description = "Active paid plan with browser automation required"), (status = 404, description = "Profile not found"), (status = 408, description = "Nothing was picked within timeout_ms; the picker has been disarmed"), - (status = 409, description = "The profile is not running, or the picker was cancelled (Escape, or a navigation)"), + (status = 409, description = "The profile is not running, the picker was cancelled (Escape, or a navigation), a native dialog holds the page (`code` DIALOG_OPEN), or the page answers nothing (`code` PAGE_NOT_RESPONDING)"), (status = 429, description = "Automation request rate limit exceeded"), (status = 502, description = "The browser did not answer as documented"), (status = 500, description = "Internal server error") @@ -5811,6 +5840,76 @@ async fn agent_pick_api( .await } +#[utoipa::path( + get, + path = "/v1/profiles/{id}/agent/dialog", + params( + ("id" = String, Path, description = "Profile ID") + ), + responses( + (status = 200, description = "Whether a native dialog (alert, confirm, prompt, \"leave this page?\") holds the page, and which. `open` is null when the page answers nothing and no dialog was seen opening; a navigation ends that", body = crate::mcp_server::PageDialog), + (status = 401, description = "Unauthorized"), + (status = 402, description = "Active paid plan with browser automation required"), + (status = 404, description = "Profile not found"), + (status = 409, description = "The profile is not running"), + (status = 502, description = "The browser did not answer as documented"), + (status = 500, description = "Internal server error") + ), + security( + ("bearer_auth" = []) + ), + tag = "agent" +)] +async fn agent_dialog_api( + Path(id): Path, +) -> Result, (StatusCode, String)> { + let ctx = agent_target_for(&id).await?; + agent_reply(&id, crate::mcp_server::agent_dialog(&ctx.target).await).await +} + +#[utoipa::path( + post, + path = "/v1/profiles/{id}/agent/handle-dialog", + params( + ("id" = String, Path, description = "Profile ID") + ), + request_body = crate::mcp_server::AgentDialogRequest, + responses( + (status = 200, description = "The dialog was accepted or dismissed. `next` is a dialog the page opened in answer, which holds the page now", body = crate::mcp_server::AgentDialogAnswer), + (status = 400, description = "Malformed request"), + (status = 401, description = "Unauthorized"), + (status = 402, description = "Active paid plan with browser automation required"), + (status = 404, description = "Profile not found"), + (status = 409, description = "The profile is not running, no dialog is open (`code` NO_DIALOG_OPEN), the page answers nothing (`code` PAGE_NOT_RESPONDING), or the person at this computer paused the agents (`code` AGENT_PAUSED) or took over the profile (`code` PROFILE_HELD_BY_HUMAN)"), + (status = 429, description = "Automation request rate limit exceeded"), + (status = 502, description = "The browser did not answer as documented"), + (status = 500, description = "Internal server error") + ), + security( + ("bearer_auth" = []) + ), + tag = "agent" +)] +async fn agent_handle_dialog_api( + Path(id): Path, + Json(request): Json, +) -> Result, (StatusCode, String)> { + if let Err(refusal) = + crate::agent_console::check_call(false, &serde_json::json!({ "profile_id": id })) + { + return Err(( + StatusCode::CONFLICT, + crate::mcp_server::McpServer::console_refusal(refusal).rest_body(), + )); + } + let ctx = agent_target_for(&id).await?; + agent_reply( + &id, + crate::mcp_server::agent_handle_dialog(&ctx.target, &request).await, + ) + .await +} + #[cfg(test)] mod url_guard_tests { //! The REST automation surface takes caller-supplied URLs too. @@ -6207,6 +6306,8 @@ mod tests { "/v1/profiles/profile-id/agent/type", "/v1/profiles/profile-id/agent/extract", "/v1/profiles/profile-id/agent/pick", + // Answering a dialog acts on the page; reading one does not. + "/v1/profiles/profile-id/agent/handle-dialog", ] { assert!( is_automation_request(&Method::POST, path), @@ -6236,6 +6337,7 @@ mod tests { (Method::POST, "/v1/profiles/profile-id/agent/unknown"), (Method::POST, "/v1/profiles/profile-id/agent/click/twice"), (Method::GET, "/v1/profiles/profile-id/agent/perceive"), + (Method::GET, "/v1/profiles/profile-id/agent/dialog"), (Method::POST, "/v1/profiles"), (Method::POST, "/v1/profiles/import"), (Method::GET, "/v1/profiles"), @@ -7085,6 +7187,25 @@ mod tests { }); assert_eq!(status, StatusCode::CONFLICT); + // A dialog holds the page: a conflict with its state, with the dialog in + // the body, the same envelope MCP answers. + let dialog = crate::page_dialog::test_support::dialog("confirm", "Delete it?"); + let (status, body) = agent_error_response(AgentError::Cdp(CdpError::DialogOpen(dialog))); + assert_eq!(status, StatusCode::CONFLICT); + let body: serde_json::Value = serde_json::from_str(&body).expect("a JSON body"); + assert_eq!(body["code"], "DIALOG_OPEN"); + assert_eq!(body["dialog"]["type"], "confirm"); + assert_eq!(body["dialog"]["message"], "Delete it?"); + assert!(body["message"].as_str().unwrap().contains("Delete it?")); + for (error, code) in [ + (AgentError::NoDialog, "NO_DIALOG_OPEN"), + (AgentError::PageNotResponding, "PAGE_NOT_RESPONDING"), + ] { + let (status, body) = agent_error_response(error); + assert_eq!(status, StatusCode::CONFLICT); + assert!(body.contains(code), "{body}"); + } + // A browser that is not there is "start it", not "we broke". let (status, _) = agent_error_response(AgentError::Cdp(CdpError::Unreachable("x".into()))); assert_eq!(status, StatusCode::CONFLICT); @@ -7692,6 +7813,7 @@ mod tests { "/v1/profiles/{id}/agent/type", "/v1/profiles/{id}/agent/extract", "/v1/profiles/{id}/agent/pick", + "/v1/profiles/{id}/agent/handle-dialog", ] { assert!( paths[path]["post"]["responses"].get("429").is_some(), @@ -7719,6 +7841,7 @@ mod tests { ("/v1/profiles/{id}/agent/type", "AgentTyping"), ("/v1/profiles/{id}/agent/extract", "Extraction"), ("/v1/profiles/{id}/agent/pick", "PickedElement"), + ("/v1/profiles/{id}/agent/handle-dialog", "AgentDialogAnswer"), ] { let reference = &paths[path]["post"]["responses"]["200"]["content"]["application/json"]["schema"]["$ref"]; @@ -7756,6 +7879,10 @@ mod tests { "AgentTypeRequest", "AgentTyping", "AgentPickRequest", + "PageDialog", + "AgentDialogRequest", + "AgentDialogAnswer", + "OpenDialog", ] { let component = &spec["components"]["schemas"][schema]; assert!( @@ -7787,6 +7914,30 @@ mod tests { serde_json::json!(["wayfern", "fallback"]) ); + // Reading a dialog is a GET that is never refused for the dialog; it is + // gated and resolved like every agent route, and is not metered. + let read = &paths["/v1/profiles/{id}/agent/dialog"]["get"]; + assert_eq!( + read["responses"]["200"]["content"]["application/json"]["schema"]["$ref"], + "#/components/schemas/PageDialog" + ); + for status in ["402", "404", "409"] { + assert!( + read["responses"].get(status).is_some(), + "GET agent/dialog is missing its {status} response" + ); + } + assert!(read["responses"].get("429").is_none()); + assert_eq!(schema_required(&spec, "AgentDialogRequest"), vec!["accept"]); + assert!(schema_required(&spec, "PageDialog").is_empty()); + let answer = schema_required(&spec, "AgentDialogAnswer"); + assert!(answer.contains(&"accepted".to_string()) && answer.contains(&"dialog".to_string())); + assert!(!answer.contains(&"next".to_string()), "next is optional"); + assert_eq!( + schema_required(&spec, "OpenDialog"), + vec!["type", "message", "default_prompt", "url"] + ); + assert!( paths["/v1/cookie-bot/runs"]["post"]["responses"] .get("429") diff --git a/src-tauri/src/cdp_target.rs b/src-tauri/src/cdp_target.rs index 53c0a3cf..017f0ffc 100644 --- a/src-tauri/src/cdp_target.rs +++ b/src-tauri/src/cdp_target.rs @@ -401,6 +401,8 @@ pub struct CdpConnection { /// page-level commands reach the page rather than the browser. cdp_session: Option, closed: Option, + /// The page a relayed connection attached to. + page: Option, /// The page's dialog state, when [`crate::page_dialog`] watches the page. dialog: Option>, } @@ -427,6 +429,7 @@ impl CdpConnection { stream, cdp_session: None, closed: None, + page: None, dialog: None, } } @@ -656,14 +659,23 @@ impl CdpConnection { let _ = self.stream.close(None).await; } - /// Move a browser-level socket onto a page. + /// Move a browser-level socket onto a page: `pinned`, the page the session's + /// tools drive, when it is known and still open, else the first page the + /// browser lists. /// /// The remote endpoint exposes `/devtools/browser/`. Every tool here /// speaks `Page.*`, `Runtime.*` and `Input.*`, which a browser socket answers /// with `'Page.navigate' wasn't found`. Attaching flat, and stamping the /// resulting session id onto everything after it, is what makes the tools /// this app already has work remotely without a single per-tool change. - async fn attach_to_page(&mut self) -> Result<(), CdpError> { + async fn attach_to_page(&mut self, pinned: Option) -> Result<(), CdpError> { + if let Some(page) = pinned { + match self.attach(&page).await { + // The page closed since it was pinned. + Err(CdpError::Protocol(_)) => {} + attached => return attached, + } + } let targets = self .call( HANDSHAKE_GET_TARGETS_ID, @@ -673,7 +685,10 @@ impl CdpConnection { .await?; let target_id = pick_remote_page_target(&targets) .ok_or_else(|| CdpError::NotDrivable("the remote browser has no page open".to_string()))?; + self.attach(&target_id).await + } + async fn attach(&mut self, target_id: &str) -> Result<(), CdpError> { let attached = self .call( HANDSHAKE_ATTACH_ID, @@ -689,6 +704,7 @@ impl CdpConnection { CdpError::Protocol("Target.attachToTarget returned no sessionId".to_string()) })?; self.cdp_session = Some(session.to_string()); + self.page = Some(target_id.to_string()); Ok(()) } } @@ -743,8 +759,8 @@ static POOL: std::sync::LazyLock< > = std::sync::LazyLock::new(Default::default); impl CdpTarget { - /// The page socket this target names: the key of its pooled connection and - /// of its dialog watch. + /// The page socket this target names, or the cloud session: the key of its + /// pooled connection. pub(crate) fn pool_key(&self) -> String { match self { Self::Local { ws_url } => ws_url.clone(), @@ -753,23 +769,40 @@ impl CdpTarget { } => format!("{ws_url}#{session_id}"), } } + + /// The key of the dialog watch of a page: the page socket here, or `page` + /// of a cloud session. + pub(crate) fn page_key(&self, page: Option<&str>) -> String { + match (self, page) { + (Self::Remote { .. }, Some(page)) => format!("{}#{page}", self.pool_key()), + _ => self.pool_key(), + } + } } /// Whether a failure on a pooled socket means the command never reached the -/// browser, so a fresh socket may send it again. +/// browser, so a fresh socket may send it again. A relayed socket whose page +/// closed answers that its page session is gone. fn not_delivered(error: &CdpError) -> bool { - matches!(error, CdpError::Transport(message) - if message.starts_with("failed to send CDP command") - || message.contains("connection closed")) + match error { + CdpError::Transport(message) => { + message.starts_with("failed to send CDP command") || message.contains("connection closed") + } + CdpError::Protocol(message) => message.contains(SESSION_GONE), + _ => false, + } } +/// How a browser refuses a command to a page session that has ended. +const SESSION_GONE: &str = "Session with given id not found"; + /// Whether the socket is still fit for the next command after `result`. A /// late reply to a command cut short by a dialog is skipped by its id. fn reusable(result: &Result) -> bool { - matches!( - result, - Ok(_) | Err(CdpError::Protocol(_)) | Err(CdpError::DialogOpen(_)) - ) + match result { + Err(CdpError::Protocol(message)) => !message.contains(SESSION_GONE), + other => matches!(other, Ok(_) | Err(CdpError::DialogOpen(_))), + } } async fn give_back(key: String, pooled: PooledConnection) { @@ -995,11 +1028,22 @@ impl CdpTarget { /// Retries a connection that failed for a reason a retry could fix, and never /// one that failed because the answer was no. pub async fn connect(&self) -> Result { + self.connect_with(true).await + } + + /// A socket to the browser of a cloud session, on no page. On this computer, + /// the page socket. + pub(crate) async fn connect_browser(&self) -> Result { + self.connect_with(false).await + } + + async fn connect_with(&self, attach: bool) -> Result { let mut attempt = 0u32; loop { - let error = match self.connect_once().await { + let error = match self.connect_once(attach).await { Ok(mut connection) => { - connection.dialog = crate::page_dialog::state_of(&self.pool_key()); + connection.dialog = + crate::page_dialog::state_of(&self.page_key(connection.page.as_deref())); return Ok(connection); } Err(e) => e, @@ -1018,7 +1062,7 @@ impl CdpTarget { } } - async fn connect_once(&self) -> Result { + async fn connect_once(&self, attach: bool) -> Result { match self { Self::Local { ws_url } => { let request = ws_url @@ -1033,7 +1077,11 @@ impl CdpTarget { session_id, } => { let mut connection = dial_relay(ws_url, bearer).await?; - if let Err(e) = connection.attach_to_page().await { + if !attach { + return Ok(connection); + } + let pinned = crate::page_dialog::remote_page(self); + if let Err(e) = connection.attach_to_page(pinned).await { log::warn!( "Remote session page attach failed session={} err=\"{e}\"", ShortId(session_id) @@ -1208,6 +1256,19 @@ pub fn classify_handshake_status(status: u16) -> CdpError { mod tests { use super::*; + #[test] + fn a_kept_socket_whose_page_closed_sends_again_on_a_new_one() { + let gone = CdpError::Protocol( + r#"{"code":-32001,"message":"Session with given id not found."}"#.to_string(), + ); + assert!(not_delivered(&gone)); + assert!(!reusable(&Err(gone))); + let refused = + CdpError::Protocol(r#"{"code":-32000,"message":"Cannot find context"}"#.to_string()); + assert!(!not_delivered(&refused)); + assert!(reusable(&Err(refused))); + } + #[test] fn a_page_target_is_preferred_over_the_devtools_frontend() { // Attaching to devtools:// drives the inspector, not the site, and the diff --git a/src-tauri/src/mcp_server.rs b/src-tauri/src/mcp_server.rs index 981c539a..78ddb0f6 100644 --- a/src-tauri/src/mcp_server.rs +++ b/src-tauri/src/mcp_server.rs @@ -400,6 +400,16 @@ pub struct McpError { data: Option, } +impl McpError { + /// The body a REST endpoint answers with for the same refusal: the + /// structured detail, plus `message`. + pub(crate) fn rest_body(&self) -> String { + let mut body = self.data.clone().unwrap_or_else(|| serde_json::json!({})); + body["message"] = serde_json::Value::from(self.message.as_str()); + body.to_string() + } +} + /// Surface a CDP failure to the agent with the reason intact. /// /// The distinction matters to whoever is on the other end: "the session is @@ -4162,6 +4172,10 @@ async fn resolve_target(profile: &BrowserProfile) -> Result /// Where a profile's browser is, whatever its page is doing. async fn locate_target(profile: &BrowserProfile) -> Result { + #[cfg(test)] + if let Some((_, target)) = pinned_profiles::get(&profile.id.to_string()) { + return Ok(target); + } crate::cdp_target::resolve(profile) .await .map_err(|e| McpError { @@ -4173,14 +4187,15 @@ async fn locate_target(profile: &BrowserProfile) -> Result /// Watch the page for dialogs, and refuse it while one holds it: a dialog that /// is open, or a page that answers nothing. `unanswered_ok` lets a navigation -/// through, which ends a dialog the watch never saw. -async fn watch_page(target: &CdpTarget, unanswered_ok: bool) -> Result<(), McpError> { +/// through, which ends a dialog the watch never saw. Every page tool and agent +/// endpoint checks this first. +pub(crate) async fn page_ready(target: &CdpTarget, unanswered_ok: bool) -> Result<(), AgentError> { use crate::page_dialog::PageState as Page; match crate::page_dialog::watch(target).await { Ok(Page::Ready) => Ok(()), - Ok(Page::Dialog(dialog)) => Err(AgentError::Cdp(CdpError::DialogOpen(dialog)).into_mcp()), + Ok(Page::Dialog(dialog)) => Err(AgentError::Cdp(CdpError::DialogOpen(dialog))), Ok(Page::Unanswered) if unanswered_ok => Ok(()), - Ok(Page::Unanswered) => Err(AgentError::PageNotResponding.into_mcp()), + Ok(Page::Unanswered) => Err(AgentError::PageNotResponding), // The tool reaches the page itself and reports what it finds there. Err(error) => { log::debug!("Page watch not opened err=\"{error}\""); @@ -4189,6 +4204,99 @@ async fn watch_page(target: &CdpTarget, unanswered_ok: bool) -> Result<(), McpEr } } +async fn watch_page(target: &CdpTarget, unanswered_ok: bool) -> Result<(), McpError> { + page_ready(target, unanswered_ok) + .await + .map_err(AgentError::into_mcp) +} + +/// What `get_dialog` answers. +#[derive(Debug, Clone, PartialEq, Serialize, ToSchema)] +pub(crate) struct PageDialog { + /// True while a dialog holds the page, false when none does, null when the + /// page answers nothing and no dialog was seen opening on it. + pub open: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dialog: Option, + /// False when the page answers nothing. + #[serde(skip_serializing_if = "Option::is_none")] + pub page_responding: Option, + /// What to do about a page that answers nothing. + #[serde(skip_serializing_if = "Option::is_none")] + pub note: Option, +} + +/// The body of `handle_dialog`. +#[derive(Debug, Clone, Deserialize, ToSchema)] +pub(crate) struct AgentDialogRequest { + /// True to accept the dialog (OK), false to dismiss it (Cancel). + pub accept: bool, + /// The text an accepted prompt dialog takes. Without it, the prompt keeps + /// its default text. + #[serde(default)] + pub prompt_text: Option, +} + +/// What `handle_dialog` did. +#[derive(Debug, Clone, Serialize, ToSchema)] +pub(crate) struct AgentDialogAnswer { + pub accepted: bool, + /// The dialog that was answered. + pub dialog: crate::page_dialog::OpenDialog, + /// A dialog the page opened in answer. It holds the page now. + #[serde(skip_serializing_if = "Option::is_none")] + pub next: Option, +} + +/// Whether a dialog holds the page `target` names, and which. Never refused: +/// it is how an agent learns why the page tools are. +pub(crate) async fn agent_dialog(target: &CdpTarget) -> Result { + use crate::page_dialog::PageState as Page; + let state = crate::page_dialog::watch(target) + .await + .map_err(AgentError::Cdp)?; + let closed = PageDialog { + open: Some(false), + dialog: None, + page_responding: None, + note: None, + }; + Ok(match state { + Page::Dialog(dialog) => PageDialog { + open: Some(true), + dialog: Some(dialog), + ..closed + }, + Page::Ready => closed, + Page::Unanswered => PageDialog { + open: None, + page_responding: Some(false), + note: Some(AgentError::PageNotResponding.message()), + ..closed + }, + }) +} + +/// Accept or dismiss the dialog that holds the page `target` names. +pub(crate) async fn agent_handle_dialog( + target: &CdpTarget, + request: &AgentDialogRequest, +) -> Result { + use crate::page_dialog::AnswerError; + let answered = crate::page_dialog::answer(target, request.accept, request.prompt_text.as_deref()) + .await + .map_err(|error| match error { + AnswerError::NoDialog => AgentError::NoDialog, + AnswerError::Unanswered => AgentError::PageNotResponding, + AnswerError::Cdp(error) => AgentError::Cdp(error), + })?; + Ok(AgentDialogAnswer { + accepted: request.accept, + dialog: answered.dialog, + next: answered.next, + }) +} + /// The tail of a page script that answers where `el` is, in viewport pixels. const RETURN_RECT_JS: &str = "const r = el.getBoundingClientRect(); return JSON.stringify({x: r.left + r.width / 2, y: r.top + r.height / 2, width: r.width, height: r.height});"; @@ -11183,6 +11291,10 @@ impl McpServer { /// the session. Whether a browser exists at all is [`resolve_cdp_target`]'s /// answer to give, because only it can see both places one could be. fn get_wayfern_profile(&self, profile_id: &str) -> Result { + #[cfg(test)] + if let Some((profile, _)) = pinned_profiles::get(profile_id) { + return Ok(profile); + } let profiles = ProfileManager::instance() .list_profiles() .map_err(|e| McpError { @@ -12178,21 +12290,11 @@ impl McpServer { ) -> Result { let profile_id = Self::require_str(arguments, "profile_id")?; let target = self.locate_cdp_target(profile_id).await?; - let state = crate::page_dialog::watch(&target) - .await - .map_err(cdp_error)?; - let answer = match state { - crate::page_dialog::PageState::Dialog(dialog) => { - serde_json::json!({ "open": true, "dialog": dialog }) - } - crate::page_dialog::PageState::Ready => serde_json::json!({ "open": false }), - crate::page_dialog::PageState::Unanswered => serde_json::json!({ - "open": serde_json::Value::Null, - "page_responding": false, - "note": AgentError::PageNotResponding.message(), - }), - }; - Self::json_content(&answer) + let dialog = agent_dialog(&target).await.map_err(|error| match error { + AgentError::Cdp(error) => cdp_error(error), + other => other.into_mcp(), + })?; + Self::json_content(&dialog) } async fn handle_handle_dialog( @@ -12200,27 +12302,27 @@ impl McpServer { arguments: &serde_json::Value, ) -> Result { let profile_id = Self::require_str(arguments, "profile_id")?; - let accept = arguments + if arguments .get("accept") .and_then(serde_json::Value::as_bool) - .ok_or_else(|| McpError { + .is_none() + { + return Err(McpError { code: -32602, message: "Missing accept: true to accept the dialog (OK), false to dismiss it (Cancel)" .to_string(), data: None, - })?; - let prompt_text = arguments - .get("prompt_text") - .and_then(serde_json::Value::as_str); + }); + } + let request: AgentDialogRequest = Self::agent_arguments(arguments)?; let target = self.locate_cdp_target(profile_id).await?; - let answered = crate::page_dialog::answer(&target, accept, prompt_text) + let answered = agent_handle_dialog(&target, &request) .await .map_err(|error| match error { - crate::page_dialog::AnswerError::NoDialog => AgentError::NoDialog.into_mcp(), - crate::page_dialog::AnswerError::Unanswered => AgentError::PageNotResponding.into_mcp(), - crate::page_dialog::AnswerError::Cdp(error) => cdp_error(error), + AgentError::Cdp(error) => cdp_error(error), + other => other.into_mcp(), })?; - let report = answer_report(&answered.dialog, accept); + let report = answer_report(&answered.dialog, request.accept); Ok(match &answered.next { Some(next) => serde_json::json!({ "content": [{ "type": "text", "text": format!("{report}\nThe page opened another dialog. {}", dialog_refusal(next)) }] @@ -12865,7 +12967,7 @@ impl McpServer { } } - fn console_refusal(refusal: crate::agent_console::Refusal) -> McpError { + pub(crate) fn console_refusal(refusal: crate::agent_console::Refusal) -> McpError { match refusal { crate::agent_console::Refusal::Paused { note } => McpError { code: -32000, @@ -13876,6 +13978,38 @@ fn cloud_failure_sentence(code: &str) -> Option<&'static str> { #[path = "mcp_server_browser_tests.rs"] mod browser_tests; +#[cfg(test)] +#[path = "mcp_server_fleet_tests.rs"] +mod fleet_tests; + +/// Profiles a test drives without a profile on disk: the tools find the +/// profile and its browser here first. +#[cfg(test)] +pub(crate) mod pinned_profiles { + use super::{BrowserProfile, CdpTarget}; + use std::collections::HashMap; + use std::sync::Mutex; + + static PINNED: Mutex>> = Mutex::new(None); + + pub(crate) fn pin(profile: BrowserProfile, target: CdpTarget) { + let mut pinned = PINNED.lock().unwrap(); + pinned + .get_or_insert_with(HashMap::new) + .insert(profile.id.to_string(), (profile, target)); + } + + pub(crate) fn unpin(profile_id: &str) { + if let Some(pinned) = PINNED.lock().unwrap().as_mut() { + pinned.remove(profile_id); + } + } + + pub(crate) fn get(profile_id: &str) -> Option<(BrowserProfile, CdpTarget)> { + PINNED.lock().unwrap().as_ref()?.get(profile_id).cloned() + } +} + #[cfg(test)] mod tests { use super::*; @@ -16524,6 +16658,51 @@ mod tests { ); } + #[tokio::test] + async fn a_dialog_is_read_the_same_way_by_every_front_door() { + use crate::page_dialog::{test_support, PageState as Page}; + use crate::wayfern_cdp::test_support::{fake_browser, Fake}; + + let (target, _) = fake_browser(Fake::Cooperative).await; + let read = |state: Page| { + let target = target.clone(); + async move { + test_support::hold(&target, state); + let read = serde_json::to_value(agent_dialog(&target).await.unwrap()).unwrap(); + let ready = page_ready(&target, false).await.err().map(|e| e.code()); + let navigable = page_ready(&target, true).await.err().map(|e| e.code()); + test_support::release(&target); + (read, ready, navigable) + } + }; + assert_eq!( + read(Page::Ready).await, + (serde_json::json!({ "open": false }), None, None) + ); + let (open, ready, navigable) = + read(Page::Dialog(test_support::dialog("prompt", "Your name?"))).await; + assert_eq!(open["open"], true); + assert_eq!(open["dialog"]["type"], "prompt"); + assert_eq!(open["dialog"]["message"], "Your name?"); + assert_eq!( + (ready, navigable), + (Some("DIALOG_OPEN"), Some("DIALOG_OPEN")) + ); + let (silent, ready, navigable) = read(Page::Unanswered).await; + assert_eq!(silent["open"], serde_json::Value::Null); + assert_eq!(silent["page_responding"], false); + assert!(silent["note"].is_string()); + assert_eq!((ready, navigable), (Some("PAGE_NOT_RESPONDING"), None)); + + let refused = McpServer::console_refusal(crate::agent_console::Refusal::Paused { + note: Some("lunch".into()), + }) + .rest_body(); + let refused: serde_json::Value = serde_json::from_str(&refused).unwrap(); + assert_eq!(refused["code"], "AGENT_PAUSED"); + assert!(refused["message"].as_str().unwrap().contains("lunch")); + } + #[test] fn an_answer_report_says_what_was_answered() { let dialog = crate::page_dialog::test_support::dialog("confirm", "Delete it?"); diff --git a/src-tauri/src/mcp_server_fleet_tests.rs b/src-tauri/src/mcp_server_fleet_tests.rs new file mode 100644 index 00000000..105d8c42 --- /dev/null +++ b/src-tauri/src/mcp_server_fleet_tests.rs @@ -0,0 +1,459 @@ +//! The dialog and typing tools on a real cloud session, reached through the +//! cloud's CDP relay the way a profile run in the cloud is. +//! +//! `DONUT_FLEET_TEST_KEY` is a `dsk_` key of a Donut service account. The test +//! makes a Linux profile named `fleet-dialogs` the first time, starts one +//! session on it and stops that session whatever happens. A session costs +//! fleet time, so the test is ignored unless asked for: +//! `cargo test --lib fleet_ -- --ignored --nocapture`. + +use super::*; +use serde_json::{json, Value}; +use std::time::{Duration, Instant}; + +const PROFILE_NAME: &str = "fleet-dialogs"; +/// The longest the session may live if the test cannot stop it. +const SESSION_CAP_SECONDS: u64 = 900; +const LIVE_WAIT: Duration = Duration::from_secs(600); + +struct Fleet { + client: reqwest::Client, + key: String, +} + +impl Fleet { + fn from_env() -> Self { + let key = std::env::var("DONUT_FLEET_TEST_KEY") + .expect("DONUT_FLEET_TEST_KEY (a dsk_ key of a Donut service account) is required"); + Self { + client: reqwest::Client::builder() + .timeout(Duration::from_secs(120)) + .build() + .unwrap(), + key, + } + } + + async fn call(&self, method: reqwest::Method, path: &str, body: Option) -> (u16, Value) { + let mut request = self + .client + .request( + method, + format!("{}{path}", crate::cloud_auth::CLOUD_API_URL), + ) + .bearer_auth(&self.key); + if let Some(body) = body { + request = request.json(&body); + } + let response = request.send().await.expect("the cloud API answers"); + let status = response.status().as_u16(); + let text = response.text().await.unwrap_or_default(); + ( + status, + serde_json::from_str(&text).unwrap_or(Value::String(text)), + ) + } + + /// The test profile, made the first time. + async fn profile_id(&self) -> String { + let (status, listed) = self + .call( + reqwest::Method::GET, + "/api/service/profiles?limit=100", + None, + ) + .await; + assert_eq!(status, 200, "{listed}"); + if let Some(id) = listed["profiles"] + .as_array() + .into_iter() + .flatten() + .find(|p| p["name"] == PROFILE_NAME) + .and_then(|p| p["id"].as_str()) + { + return id.to_string(); + } + let (status, created) = self + .call( + reqwest::Method::POST, + "/api/service/profiles", + Some(json!({ "name": PROFILE_NAME, "platform": "linux" })), + ) + .await; + assert_eq!(status, 201, "{created}"); + created["id"].as_str().unwrap().to_string() + } + + async fn start(&self, profile_id: &str) -> String { + let started = Instant::now(); + let (status, session) = self + .call( + reqwest::Method::POST, + "/api/remote-sessions", + Some(json!({ + "profile_id": profile_id, + "platform": "linux", + "idempotency_key": uuid::Uuid::new_v4().to_string(), + "max_duration_seconds": SESSION_CAP_SECONDS, + })), + ) + .await; + assert_eq!(status, 201, "{session}"); + let id = session["session_id"].as_str().unwrap().to_string(); + eprintln!("fleet: session {id} started"); + while started.elapsed() < LIVE_WAIT { + let (status, view) = self + .call( + reqwest::Method::GET, + &format!("/api/remote-sessions/{id}"), + None, + ) + .await; + if status == 200 && view["state"] == "live" && view["cdp_ready"] == true { + eprintln!("fleet: live after {} s", started.elapsed().as_secs()); + return id; + } + if status != 200 || view["state"] == "closed" || view["state"] == "error" { + self.stop(&id).await; + panic!("the session did not come up: {status} {view}"); + } + tokio::time::sleep(Duration::from_secs(3)).await; + } + self.stop(&id).await; + panic!("the session was not live within {LIVE_WAIT:?}"); + } + + async fn target(&self, session_id: &str) -> CdpTarget { + let (status, endpoint) = self + .call( + reqwest::Method::GET, + &format!("/api/remote-sessions/{session_id}/cdp"), + None, + ) + .await; + assert_eq!(status, 200, "{endpoint}"); + CdpTarget::Remote { + ws_url: endpoint["ws_url"].as_str().unwrap().to_string(), + bearer: self.key.clone(), + session_id: session_id.to_string(), + } + } + + async fn stop(&self, session_id: &str) { + let (status, stopped) = self + .call( + reqwest::Method::DELETE, + &format!("/api/remote-sessions/{session_id}"), + None, + ) + .await; + eprintln!( + "fleet: session {session_id} stopped: {status} billed_seconds={}", + stopped["billed_seconds"] + ); + } +} + +/// A button that asks to confirm, and a text field. +const FIXTURE: &str = r#"(() => { + document.title = 'Fleet dialogs'; + document.body.innerHTML = ''; + return document.visibilityState; +})()"#; + +struct Agent<'a> { + server: &'a McpServer, + profile_id: String, +} + +impl Agent<'_> { + fn args(&self, mut arguments: Value) -> Value { + arguments["profile_id"] = json!(self.profile_id); + arguments + } + + async fn evaluate(&self, expression: &str) -> Result { + self + .server + .handle_evaluate_javascript(&self.args(json!({ "expression": expression }))) + .await + .map(|result| json_text(&result)["value"].clone()) + } + + async fn navigate_and_build(&self) -> String { + self + .server + .handle_navigate(&self.args(json!({ "url": "about:blank" }))) + .await + .expect("navigate"); + self + .evaluate(FIXTURE) + .await + .expect("the fixture is built") + .as_str() + .unwrap_or_default() + .to_string() + } + + async fn get_dialog(&self) -> Value { + json_text( + &self + .server + .handle_get_dialog(&self.args(json!({}))) + .await + .expect("get_dialog"), + ) + } +} + +fn text_of(result: &Value) -> String { + result["content"][0]["text"] + .as_str() + .unwrap_or_default() + .to_string() +} + +fn json_text(result: &Value) -> Value { + serde_json::from_str(&text_of(result)).unwrap_or(Value::Null) +} + +fn code_of(error: &McpError) -> String { + error + .data + .as_ref() + .and_then(|data| data["code"].as_str()) + .unwrap_or_default() + .to_string() +} + +async fn timed(what: &str, call: impl std::future::Future) -> (T, Duration) { + let started = Instant::now(); + let value = call.await; + let elapsed = started.elapsed(); + eprintln!("fleet: {what} took {} ms", elapsed.as_millis()); + (value, elapsed) +} + +/// Run `method` on the browser itself, beside the page the tools drive. +async fn on_browser(target: &CdpTarget, method: &str, params: Value) -> Value { + let mut connection = target.connect().await.expect("a relay socket"); + connection + .send_to(None, 77, method, params) + .await + .expect("sent to the browser"); + let reply = connection + .await_reply(77, Duration::from_secs(20)) + .await + .unwrap_or_else(|e| panic!("{method}: {e}")); + connection.close().await; + reply +} + +async fn browser_version(target: &CdpTarget) -> String { + let version = on_browser(target, "Browser.getVersion", json!({})).await; + version["product"] + .as_str() + .and_then(|product| product.split('/').nth(1)) + .unwrap_or("156.0.0.0") + .to_string() +} + +async fn drive(target: CdpTarget) { + let version = browser_version(&target).await; + eprintln!("fleet: browser {version}"); + let profile = BrowserProfile { + id: uuid::Uuid::new_v4(), + name: "fleet test".to_string(), + browser: "wayfern".to_string(), + version, + ..Default::default() + }; + let profile_id = profile.id.to_string(); + pinned_profiles::pin(profile, target.clone()); + let server = McpServer::new(); + let agent = Agent { + server: &server, + profile_id: profile_id.clone(), + }; + + // a) A click that opens a confirm, and every page tool after it. + agent.navigate_and_build().await; + assert_eq!(agent.get_dialog().await, json!({ "open": false })); + let (clicked, _) = timed( + "click opening a confirm", + server.handle_click_element(&agent.args(json!({ "selector": "#delete" }))), + ) + .await; + match &clicked { + Ok(result) => eprintln!("fleet: click said {:?}", text_of(result)), + Err(error) => eprintln!( + "fleet: click refused {} {:?}", + code_of(error), + error.message + ), + } + let (refused, took) = timed("evaluate during the dialog", agent.evaluate("1 + 1")).await; + let refused = refused.expect_err("a page held by a dialog refuses the tool"); + assert_eq!(code_of(&refused), "DIALOG_OPEN", "{refused:?}"); + assert!(took < Duration::from_millis(1_700), "refused in {took:?}"); + eprintln!("fleet: agent saw {:?}", refused.message); + let (info, took) = timed( + "get_page_info during the dialog", + server.handle_get_page_info(&agent.args(json!({}))), + ) + .await; + assert_eq!(code_of(&info.expect_err("refused")), "DIALOG_OPEN"); + assert!(took < Duration::from_millis(1_700), "refused in {took:?}"); + let (read, _) = timed("get_dialog", agent.get_dialog()).await; + assert_eq!(read["open"], true, "{read}"); + assert_eq!(read["dialog"]["type"], "confirm"); + assert_eq!(read["dialog"]["message"], "Delete it?"); + let (answered, _) = timed( + "handle_dialog", + server.handle_handle_dialog(&agent.args(json!({ "accept": false }))), + ) + .await; + let answered = text_of(&answered.expect("handle_dialog")); + eprintln!("fleet: handle_dialog said {answered:?}"); + assert!( + answered.starts_with("Dismissed a confirm dialog that says \"Delete it?\"."), + "{answered}" + ); + assert_eq!( + agent + .evaluate("document.body.dataset.deleted") + .await + .unwrap(), + "false" + ); + let none = server + .handle_handle_dialog(&agent.args(json!({ "accept": true }))) + .await + .expect_err("nothing to answer"); + assert_eq!(code_of(&none), "NO_DIALOG_OPEN"); + + // b) Another tab comes to the front and the agent's tab loads again behind + // it: the text still reaches the field. + let front = on_browser( + &target, + "Target.createTarget", + json!({ "url": "about:blank" }), + ) + .await; + let front = front["targetId"].as_str().unwrap().to_string(); + let visibility = agent.navigate_and_build().await; + assert_eq!( + visibility, "hidden", + "the agent's tab loaded behind the other" + ); + let (typed, _) = timed( + "type_locator in the tab behind", + server.handle_type_locator(&agent.args(json!({ + "locator": { "role": "textbox", "name": "Later" }, + "text": "typed behind", + }))), + ) + .await; + let typed = json_text(&typed.expect("typed")); + eprintln!("fleet: typed_with={} {typed}", typed["typed_with"]); + assert_eq!(typed["verified"], true, "{typed}"); + assert!( + typed["typed_with"] == "key_events" || typed["typed_with"] == "browser_keyboard", + "{typed}" + ); + assert_eq!( + agent + .evaluate("document.querySelector('#later').value") + .await + .unwrap(), + "typed behind" + ); + let (typed, _) = timed( + "type_text in the tab behind", + server.handle_type_text(&agent.args(json!({ + "selector": "#later", + "text": "by selector", + }))), + ) + .await; + eprintln!( + "fleet: type_text said {:?}", + text_of(&typed.expect("typed")) + ); + assert_eq!( + agent + .evaluate("document.querySelector('#later').value") + .await + .unwrap(), + "by selector" + ); + on_browser(&target, "Target.closeTarget", json!({ "targetId": front })).await; + + // c) The agent's tab closes while a dialog holds it: the tab that is left + // is driven, no dialog is reported on it, and its own dialogs are seen. + agent + .evaluate("setTimeout(() => alert('Closing'), 50); 1") + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(800)).await; + assert_eq!(agent.get_dialog().await["dialog"]["message"], "Closing"); + let agents_tab = crate::page_dialog::remote_page(&target).expect("a pinned tab"); + on_browser( + &target, + "Target.createTarget", + json!({ "url": "about:blank" }), + ) + .await; + on_browser( + &target, + "Target.closeTarget", + json!({ "targetId": agents_tab }), + ) + .await; + tokio::time::sleep(Duration::from_millis(500)).await; + let (after, _) = timed("get_dialog after the tab closed", agent.get_dialog()).await; + assert_eq!(after, json!({ "open": false }), "{after}"); + let (title, _) = timed("evaluate on the tab left", agent.evaluate("1 + 1")).await; + assert_eq!(title.expect("the tab left is driven"), 2); + assert_ne!( + crate::page_dialog::remote_page(&target).as_deref(), + Some(agents_tab.as_str()) + ); + agent + .evaluate("setTimeout(() => confirm('Again?'), 50); 1") + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(800)).await; + let (refused, took) = timed( + "evaluate during a dialog on the tab left", + agent.evaluate("1 + 1"), + ) + .await; + assert_eq!( + code_of(&refused.expect_err("refused")), + "DIALOG_OPEN", + "the tab left is watched" + ); + assert!(took < Duration::from_millis(1_700), "refused in {took:?}"); + server + .handle_handle_dialog(&agent.args(json!({ "accept": true }))) + .await + .expect("the dialog on the tab left is answered"); + assert_eq!(agent.evaluate("1 + 1").await.expect("works again"), 2); + + pinned_profiles::unpin(&profile_id); +} + +#[tokio::test(flavor = "multi_thread")] +#[ignore = "starts a cloud session: needs DONUT_FLEET_TEST_KEY"] +async fn fleet_dialogs_and_typing_behind_another_tab() { + let fleet = Fleet::from_env(); + let profile_id = fleet.profile_id().await; + let session = fleet.start(&profile_id).await; + let target = fleet.target(&session).await; + let outcome = tokio::spawn(drive(target)).await; + fleet.stop(&session).await; + if let Err(failure) = outcome { + std::panic::resume_unwind(failure.into_panic()); + } +} diff --git a/src-tauri/src/page_dialog.rs b/src-tauri/src/page_dialog.rs index c6268816..dbfc5349 100644 --- a/src-tauri/src/page_dialog.rs +++ b/src-tauri/src/page_dialog.rs @@ -3,25 +3,33 @@ //! //! While such a dialog is open the page runs no script and answers no command //! that needs it. Only a CDP session whose Page domain was on when the dialog -//! opened can answer it; a session opened later cannot even turn the domain on. -//! So the first time a tool touches a page, a socket with the Page domain on is -//! opened to it and kept for as long as the page lives. It sees each dialog open -//! and close, answers one when asked, and lets every other connection to the -//! page fail at once instead of waiting for a reply that cannot come. +//! opened can answer it, and a session opened later gets its `Page.enable` +//! answered only once the dialog closes. So the first time a tool touches a +//! page, a session with the Page domain on is opened to it and kept for as long +//! as the page lives. It sees each dialog open and close, answers one when +//! asked, and lets every other connection to the page fail at once instead of +//! waiting for a reply that cannot come. +//! +//! A page on this computer gets a socket of its own. A cloud session gets one +//! relay socket for all its pages: it follows the session's tabs and pins the +//! one the tools drive, so every connection to the session attaches to the +//! page that is watched, and the watch costs one of the session's few relay +//! attachments however many tabs it has. -use crate::cdp_target::{cdp_frame, CdpError, CdpTarget}; +use crate::cdp_target::{cdp_frame, pick_remote_page_target, CdpError, CdpTarget}; use futures_util::sink::SinkExt; use futures_util::stream::StreamExt; use serde::Serialize; use serde_json::Value; use std::collections::HashMap; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, LazyLock, Mutex}; use std::time::Duration; use tokio::sync::{mpsc, oneshot, watch}; use tokio_tungstenite::tungstenite::Message; -/// How long a new watch waits for the page to turn its Page domain on. +/// How long a new watch waits for the page to turn its Page domain on, and a +/// cloud session's browser to list or attach its pages. const ENABLE_WAIT: Duration = Duration::from_secs(3); /// How long one `Page.enable` of a page that does not answer is waited for /// before another is sent. @@ -49,7 +57,7 @@ pub fn answered_during_dialog(method: &str) -> bool { } /// A dialog open on a page. -#[derive(Debug, Clone, PartialEq, Serialize)] +#[derive(Debug, Clone, PartialEq, Serialize, utoipa::ToSchema)] pub struct OpenDialog { /// `alert`, `confirm`, `prompt` or `beforeunload`. #[serde(rename = "type")] @@ -155,31 +163,156 @@ impl PageState { type Replies = Mutex>>>; -/// The kept socket to one page. -pub struct PageWatch { +/// A kept socket: to one page on this computer, or to the browser of a cloud +/// session, where each watched page is a CDP session of its own. +struct Link { serial: u64, outgoing: mpsc::UnboundedSender, - session: Option, replies: Replies, - state: watch::Sender, next_id: AtomicU64, + /// A cloud session's targets, in the order the browser reported them. + targets: Mutex>, + /// The page of a cloud session the tools drive. + pinned: Mutex>, +} + +static LINK_SERIAL: AtomicU64 = AtomicU64::new(1); + +impl Link { + fn new(outgoing: mpsc::UnboundedSender) -> Self { + Self { + serial: LINK_SERIAL.fetch_add(1, Ordering::Relaxed), + outgoing, + replies: Mutex::default(), + next_id: AtomicU64::new(1), + targets: Mutex::default(), + pinned: Mutex::default(), + } + } + + fn send( + &self, + session: Option<&str>, + id: u64, + method: &str, + params: Value, + ) -> Result<(), CdpError> { + let frame = cdp_frame(session, id, method, params); + self + .outgoing + .send(Message::Text(frame.to_string().into())) + .map_err(|_| CdpError::Transport("the page's watch socket is closed".to_string())) + } + + async fn call( + &self, + session: Option<&str>, + method: &str, + params: Value, + timeout: Duration, + ) -> Result { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = oneshot::channel(); + lock(&self.replies).insert(id, tx); + if let Err(error) = self.send(session, id, method, params) { + lock(&self.replies).remove(&id); + return Err(error); + } + match tokio::time::timeout(timeout, rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(CdpError::Transport( + "the page closed before it answered".to_string(), + )), + Err(_) => { + lock(&self.replies).remove(&id); + Err(CdpError::Transport( + "timed out waiting for a CDP response".to_string(), + )) + } + } + } + + /// Hand a reply to the caller that waits for it. False for an event. + fn reply(&self, message: &Value) -> bool { + let Some(id) = message.get("id").and_then(Value::as_u64) else { + return false; + }; + if let Some(waiter) = lock(&self.replies).remove(&id) { + let reply = match message.get("error") { + Some(error) => Err(CdpError::Protocol(error.to_string())), + None => Ok(message.get("result").cloned().unwrap_or_default()), + }; + let _ = waiter.send(reply); + } + true + } + + fn close(&self) { + let _ = self.outgoing.send(Message::Close(None)); + } + + fn note_target(&self, info: &Value) { + let Some(id) = info.get("targetId").and_then(Value::as_str) else { + return; + }; + let mut targets = lock(&self.targets); + match targets.iter_mut().find(|known| known["targetId"] == id) { + Some(known) => *known = info.clone(), + None => targets.push(info.clone()), + } + } + + fn forget_target(&self, id: &str) { + lock(&self.targets).retain(|known| known["targetId"] != id); + } + + /// The page the tools drive: the pinned one while it is open, else the + /// first page the browser lists, which is pinned from then on. The browser + /// lists its tabs in no fixed order, so without the pin each new connection + /// could drive another tab. + fn page(&self) -> Option { + let targets = lock(&self.targets); + let mut pinned = lock(&self.pinned); + if let Some(page) = pinned.as_deref() { + if targets.iter().any(|known| known["targetId"] == page) { + return pinned.clone(); + } + } + *pinned = pick_remote_page_target(&serde_json::json!({ "targetInfos": *targets })); + pinned.clone() + } +} + +/// What the watch of one page knows. +pub struct PageWatch { + link: Arc, + /// The page's CDP session on a cloud session's socket. + session: Option, + /// The page's target id on a cloud session's socket. + page: Option, + state: watch::Sender, /// A `Page.enable` is waiting for its answer. - enabling: std::sync::atomic::AtomicBool, + enabling: AtomicBool, } impl PageWatch { - fn new(outgoing: mpsc::UnboundedSender, session: Option) -> Self { + fn new(link: Arc, session: Option, page: Option) -> Self { Self { - serial: WATCH_SERIAL.fetch_add(1, Ordering::Relaxed), - outgoing, + link, session, - replies: Mutex::default(), + page, state: watch::Sender::new(PageState::Unanswered), - next_id: AtomicU64::new(1), - enabling: std::sync::atomic::AtomicBool::new(false), + enabling: AtomicBool::new(false), } } + async fn call(&self, method: &str, params: Value, timeout: Duration) -> Result { + self + .link + .call(self.session.as_deref(), method, params, timeout) + .await + } + /// Turn the Page domain on, unless an earlier attempt still waits. A page /// held by a dialog answers once something ends the dialog; an attempt that /// is never answered is followed by another on the next look. @@ -206,55 +339,16 @@ impl PageWatch { }); } - fn send(&self, id: u64, method: &str, params: Value) -> Result<(), CdpError> { - let frame = cdp_frame(self.session.as_deref(), id, method, params); - self - .outgoing - .send(Message::Text(frame.to_string().into())) - .map_err(|_| CdpError::Transport("the page's watch socket is closed".to_string())) - } - - async fn call(&self, method: &str, params: Value, timeout: Duration) -> Result { - let id = self.next_id.fetch_add(1, Ordering::Relaxed); - let (tx, rx) = oneshot::channel(); - lock(&self.replies).insert(id, tx); - if let Err(error) = self.send(id, method, params) { - lock(&self.replies).remove(&id); - return Err(error); - } - match tokio::time::timeout(timeout, rx).await { - Ok(Ok(result)) => result, - Ok(Err(_)) => Err(CdpError::Transport( - "the page closed before it answered".to_string(), - )), - Err(_) => { - lock(&self.replies).remove(&id); - Err(CdpError::Transport( - "timed out waiting for a CDP response".to_string(), - )) - } - } - } - - /// Read one message off the page and update what is known. - fn take(&self, message: &Value) { - if let Some(id) = message.get("id").and_then(Value::as_u64) { - if let Some(waiter) = lock(&self.replies).remove(&id) { - let reply = match message.get("error") { - Some(error) => Err(CdpError::Protocol(error.to_string())), - None => Ok(message.get("result").cloned().unwrap_or_default()), - }; - let _ = waiter.send(reply); - } - return; - } - match message.get("method").and_then(Value::as_str) { - Some("Page.javascriptDialogOpening") => { - let dialog = OpenDialog::from_event(message.get("params").unwrap_or(&Value::Null)); + /// A dialog event of this page. A browser may announce a dialog that opened + /// before the watch, in answer to its `Page.enable`. + fn take_dialog(&self, method: &str, params: &Value) { + match method { + "Page.javascriptDialogOpening" => { + let dialog = OpenDialog::from_event(params); log::debug!("Page dialog opened type={}", dialog.kind); self.state.send_replace(PageState::Dialog(dialog)); } - Some("Page.javascriptDialogClosed") => { + "Page.javascriptDialogClosed" => { log::debug!("Page dialog closed"); self.state.send_replace(PageState::Ready); } @@ -263,12 +357,14 @@ impl PageWatch { } } -static WATCH_SERIAL: AtomicU64 = AtomicU64::new(1); - +/// Watched pages, by [`CdpTarget::page_key`]. static WATCHES: LazyLock>>> = LazyLock::new(Default::default); -/// One watch is opened at a time, so two tools that touch a new page together -/// do not open two. +/// The sockets of cloud sessions, by [`CdpTarget::pool_key`]. +static LINKS: LazyLock>>> = LazyLock::new(Default::default); + +/// One socket or page session is opened at a time, so two tools that touch a +/// new page together do not open two. static OPENING: LazyLock> = LazyLock::new(Default::default); fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { @@ -281,18 +377,125 @@ fn registered(key: &str) -> Option> { lock(&WATCHES).get(key).cloned() } -/// The dialog state of the page behind `key` (a target's pool key), when the -/// page is watched. +/// The watch that `session` names on `link`. +fn watch_on(link: &Link, session: Option<&str>) -> Option> { + lock(&WATCHES) + .values() + .find(|watch| watch.link.serial == link.serial && watch.session.as_deref() == session) + .cloned() +} + +/// Stop watching the pages of `link` that `gone` names. Their connections +/// see the state end and stop counting on it. +fn drop_watches(link: &Link, gone: impl Fn(&PageWatch) -> bool) { + lock(&WATCHES).retain(|_, watch| watch.link.serial != link.serial || !gone(watch)); +} + +/// Read one message off a link and update what is known. +fn take(link: &Link, message: &Value) { + if link.reply(message) { + return; + } + let params = message.get("params").unwrap_or(&Value::Null); + match message.get("method").and_then(Value::as_str) { + Some(method @ ("Page.javascriptDialogOpening" | "Page.javascriptDialogClosed")) => { + let session = message.get("sessionId").and_then(Value::as_str); + if let Some(watch) = watch_on(link, session) { + watch.take_dialog(method, params); + } + } + Some("Target.targetCreated" | "Target.targetInfoChanged") => { + link.note_target(¶ms["targetInfo"]); + } + Some("Target.targetDestroyed") => { + if let Some(page) = params["targetId"].as_str() { + link.forget_target(page); + drop_watches(link, |watch| watch.page.as_deref() == Some(page)); + } + } + Some("Target.detachedFromTarget") => { + if let Some(session) = params["sessionId"].as_str() { + drop_watches(link, |watch| watch.session.as_deref() == Some(session)); + } + } + _ => {} + } +} + +/// Keep `stream` open as a link, read for as long as it lives. +fn open_link(stream: crate::cdp_target::RelaySocket) -> Arc { + let (mut sink, mut stream) = stream.split(); + let (outgoing, mut queued) = mpsc::unbounded_channel::(); + let link = Arc::new(Link::new(outgoing)); + tokio::spawn(async move { + while let Some(message) = queued.recv().await { + let closing = matches!(message, Message::Close(_)); + if sink.send(message).await.is_err() || closing { + break; + } + } + let _ = sink.close().await; + }); + let reader = link.clone(); + tokio::spawn(async move { + while let Some(frame) = stream.next().await { + match frame { + Ok(Message::Text(text)) => { + if let Ok(message) = serde_json::from_str::(&text) { + take(&reader, &message); + } + } + Ok(Message::Close(_)) | Err(_) => break, + Ok(_) => {} + } + } + log::debug!("Page watch socket closed"); + lock(&LINKS).retain(|_, kept| kept.serial != reader.serial); + drop_watches(&reader, |_| true); + lock(&reader.replies).clear(); + }); + link +} + +/// The dialog state of the page behind `key` ([`CdpTarget::page_key`]), when +/// the page is watched. pub fn state_of(key: &str) -> Option> { registered(key).map(|watch| watch.state.subscribe()) } +/// The page of a cloud session the tools drive, once the session's socket is +/// open: every connection to the session attaches to it. +pub fn remote_page(target: &CdpTarget) -> Option { + if !target.is_remote() { + return None; + } + let link = lock(&LINKS).get(&target.pool_key()).cloned()?; + link.page() +} + +/// The key of the page the tools on `target` drive now. +fn current_key(target: &CdpTarget) -> Option { + if target.is_remote() { + Some(target.page_key(Some(&remote_page(target)?))) + } else { + Some(target.pool_key()) + } +} + /// The dialog open on the page `target` names, when it is watched and one is. pub fn open_dialog(target: &CdpTarget) -> Option { - registered(&target.pool_key()).and_then(|watch| watch.state.borrow().dialog().cloned()) + registered(¤t_key(target)?).and_then(|watch| watch.state.borrow().dialog().cloned()) } async fn ensure(target: &CdpTarget) -> Result, CdpError> { + if target.is_remote() { + ensure_remote(target).await + } else { + ensure_local(target).await + } +} + +async fn ensure_local(target: &CdpTarget) -> Result, CdpError> { let key = target.pool_key(); if let Some(watch) = registered(&key) { return Ok(watch); @@ -301,38 +504,70 @@ async fn ensure(target: &CdpTarget) -> Result, CdpError> { if let Some(watch) = registered(&key) { return Ok(watch); } - let (stream, session) = target.connect().await?.into_parts(); - let (mut sink, mut stream) = stream.split(); - let (outgoing, mut queued) = mpsc::unbounded_channel::(); - let watch = Arc::new(PageWatch::new(outgoing, session)); - lock(&WATCHES).insert(key.clone(), watch.clone()); + let (stream, _) = target.connect().await?.into_parts(); + let watch = Arc::new(PageWatch::new(open_link(stream), None, None)); + lock(&WATCHES).insert(key, watch.clone()); + watch.enable(); + Ok(watch) +} - tokio::spawn(async move { - while let Some(message) = queued.recv().await { - if sink.send(message).await.is_err() { - break; - } - } - let _ = sink.close().await; - }); - let reader = watch.clone(); - tokio::spawn(async move { - while let Some(frame) = stream.next().await { - match frame { - Ok(Message::Text(text)) => { - if let Ok(message) = serde_json::from_str::(&text) { - reader.take(&message); - } - } - Ok(Message::Close(_)) | Err(_) => break, - Ok(_) => {} - } - } - log::debug!("Page watch closed"); - lock(&WATCHES).retain(|_, kept| kept.serial != reader.serial); - lock(&reader.replies).clear(); - }); +/// The socket of a cloud session, opened and told to report the session's +/// targets the first time. The browser lists every target it has before it +/// answers, then reports each one that opens, changes or closes. +async fn link_of(target: &CdpTarget) -> Result, CdpError> { + let key = target.pool_key(); + if let Some(link) = lock(&LINKS).get(&key).cloned() { + return Ok(link); + } + let _opening = OPENING.lock().await; + if let Some(link) = lock(&LINKS).get(&key).cloned() { + return Ok(link); + } + let (stream, _) = target.connect_browser().await?.into_parts(); + let link = open_link(stream); + if let Err(error) = link + .call( + None, + "Target.setDiscoverTargets", + serde_json::json!({ "discover": true }), + ENABLE_WAIT, + ) + .await + { + link.close(); + return Err(error); + } + lock(&LINKS).insert(key, link.clone()); + Ok(link) +} +async fn ensure_remote(target: &CdpTarget) -> Result, CdpError> { + let link = link_of(target).await?; + let page = link + .page() + .ok_or_else(|| CdpError::NotDrivable("the remote browser has no page open".to_string()))?; + let key = target.page_key(Some(&page)); + if let Some(watch) = registered(&key) { + return Ok(watch); + } + let _opening = OPENING.lock().await; + if let Some(watch) = registered(&key) { + return Ok(watch); + } + let attached = link + .call( + None, + "Target.attachToTarget", + serde_json::json!({ "targetId": page, "flatten": true }), + ENABLE_WAIT, + ) + .await?; + let session = attached + .get("sessionId") + .and_then(Value::as_str) + .ok_or_else(|| CdpError::Protocol("Target.attachToTarget returned no sessionId".to_string()))?; + let watch = Arc::new(PageWatch::new(link, Some(session.to_string()), Some(page))); + lock(&WATCHES).insert(key, watch.clone()); watch.enable(); Ok(watch) } @@ -419,7 +654,7 @@ pub(crate) mod test_support { /// connections to `target` behave as on a page in that state. pub(crate) fn hold(target: &CdpTarget, state: PageState) -> Arc { let (outgoing, _) = mpsc::unbounded_channel(); - let watch = Arc::new(PageWatch::new(outgoing, None)); + let watch = Arc::new(PageWatch::new(Arc::new(Link::new(outgoing)), None, None)); watch.state.send_replace(state); lock(&WATCHES).insert(target.pool_key(), watch.clone()); watch @@ -445,12 +680,17 @@ mod tests { #[test] fn a_dialog_event_is_read_into_the_state_and_its_close_clears_it() { let (outgoing, _) = mpsc::unbounded_channel(); - let watch = PageWatch::new(outgoing, None); + let link = Arc::new(Link::new(outgoing)); + let watch = Arc::new(PageWatch::new(link.clone(), None, None)); + lock(&WATCHES).insert("read-test".to_string(), watch.clone()); watch.state.send_replace(PageState::Ready); - watch.take(&serde_json::json!({ - "method": "Page.javascriptDialogOpening", - "params": { "type": "prompt", "message": "Your name?", "defaultPrompt": "Ann", "url": "http://a.test/", "hasBrowserHandler": true } - })); + take( + &link, + &serde_json::json!({ + "method": "Page.javascriptDialogOpening", + "params": { "type": "prompt", "message": "Your name?", "defaultPrompt": "Ann", "url": "http://a.test/", "hasBrowserHandler": true } + }), + ); let open = watch .state .borrow() @@ -465,10 +705,12 @@ mod tests { ), ("prompt", "Your name?", "Ann") ); - watch.take( + take( + &link, &serde_json::json!({ "method": "Page.javascriptDialogClosed", "params": { "result": true } }), ); assert_eq!(*watch.state.borrow(), PageState::Ready); + lock(&WATCHES).remove("read-test"); } #[test] @@ -733,7 +975,9 @@ mod tests { watch(&target).await.unwrap(); let watch = registered(&target.pool_key()).unwrap(); watch + .link .send( + None, 99, "Runtime.evaluate", serde_json::json!({ "expression": "close" }), @@ -748,4 +992,507 @@ mod tests { .await; assert!(gone.is_ok()); } + + /// A browser behind page sockets and relay sockets: pages, flat sessions, + /// target discovery, and alerts that hold their page. A command to a page + /// an alert holds is never answered. With `reannounce`, a session that turns + /// the Page domain on while an alert is open is told of the alert at once; + /// its `Page.enable` is answered when the alert closes, in either case. + #[derive(Default)] + struct Browser { + /// Pages, in the order `Target.getTargets` lists them. + pages: Vec, + /// The message of the alert open on a page. + alerts: HashMap, + sockets: Vec, + reannounce: bool, + /// Every request, with the socket it came on. + requests: Vec<(usize, Value)>, + next_page: u32, + } + + struct FakeSocket { + out: mpsc::UnboundedSender>, + /// The page a page socket is on. + page: Option, + /// Flat sessions and their pages. + sessions: HashMap, + /// Sessions with the Page domain on; "" is the page socket itself. + enabled: Vec, + /// `Page.enable` answers an alert holds back: (session, id). + held: Vec<(String, Value)>, + discover: bool, + } + + fn frame(session: &str, body: Value) -> Value { + let mut body = body; + if !session.is_empty() { + body["sessionId"] = Value::from(session); + } + body + } + + impl Browser { + fn send(&self, socket: usize, frame: Value) { + let _ = self.sockets[socket].out.send(Some(frame.to_string())); + } + + fn info(page: &str) -> Value { + serde_json::json!({ "targetId": page, "type": "page", "url": format!("http://{page}.test/"), "attached": false }) + } + + /// Send `method` to every session with the Page domain on `page`. + fn to_page(&self, page: &str, method: &str, params: Value) { + for (index, socket) in self.sockets.iter().enumerate() { + for key in &socket.enabled { + let on_page = if key.is_empty() { + socket.page.as_deref() == Some(page) + } else { + socket.sessions.get(key).map(String::as_str) == Some(page) + }; + if on_page { + self.send( + index, + frame( + key, + serde_json::json!({ "method": method, "params": params }), + ), + ); + } + } + } + } + + fn alert_event(message: &str) -> Value { + serde_json::json!({ "type": "alert", "message": message, "defaultPrompt": "", "url": "http://a.test/" }) + } + + fn open_alert(&mut self, page: &str, message: &str) { + self.alerts.insert(page.to_string(), message.to_string()); + self.to_page( + page, + "Page.javascriptDialogOpening", + Self::alert_event(message), + ); + } + + fn close_alert(&mut self, page: &str) { + if self.alerts.remove(page).is_none() { + return; + } + self.to_page( + page, + "Page.javascriptDialogClosed", + serde_json::json!({ "result": true, "userInput": "" }), + ); + for index in 0..self.sockets.len() { + let socket = &mut self.sockets[index]; + let page_socket = socket.page.clone(); + let sessions = socket.sessions.clone(); + let (released, kept): (Vec<_>, Vec<_>) = socket.held.drain(..).partition(|(key, _)| { + if key.is_empty() { + page_socket.as_deref() == Some(page) + } else { + sessions.get(key).map(String::as_str) == Some(page) + } + }); + socket.held = kept; + for (key, id) in released { + self.send( + index, + frame(&key, serde_json::json!({ "id": id, "result": {} })), + ); + } + } + } + + fn close_page(&mut self, page: &str) { + self.close_alert(page); + self.pages.retain(|known| known != page); + for index in 0..self.sockets.len() { + let gone: Vec = self.sockets[index] + .sessions + .iter() + .filter(|(_, on)| on.as_str() == page) + .map(|(session, _)| session.clone()) + .collect(); + for session in gone { + self.sockets[index].sessions.remove(&session); + self.send( + index, + serde_json::json!({ "method": "Target.detachedFromTarget", "params": { "sessionId": session, "targetId": page } }), + ); + } + if self.sockets[index].discover { + self.send( + index, + serde_json::json!({ "method": "Target.targetDestroyed", "params": { "targetId": page } }), + ); + } + if self.sockets[index].page.as_deref() == Some(page) { + let _ = self.sockets[index].out.send(None); + } + } + } + + fn handle(&mut self, socket: usize, request: Value) { + self.requests.push((socket, request.clone())); + let id = request["id"].clone(); + let session = request["sessionId"] + .as_str() + .unwrap_or_default() + .to_string(); + let params = &request["params"]; + let reply = + |result: Value| frame(&session, serde_json::json!({ "id": id, "result": result })); + let page = if session.is_empty() { + self.sockets[socket].page.clone() + } else { + match self.sockets[socket].sessions.get(&session) { + Some(page) => Some(page.clone()), + None => { + self.send( + socket, + frame( + &session, + serde_json::json!({ "id": id, "error": { "code": -32001, "message": "Session with given id not found." } }), + ), + ); + return; + } + } + }; + match request["method"].as_str().unwrap_or_default() { + "Target.getTargets" => { + let infos: Vec = self.pages.iter().map(|page| Self::info(page)).collect(); + self.send(socket, reply(serde_json::json!({ "targetInfos": infos }))); + } + "Target.setDiscoverTargets" => { + for page in self.pages.clone() { + self.send( + socket, + serde_json::json!({ "method": "Target.targetCreated", "params": { "targetInfo": Self::info(&page) } }), + ); + } + self.sockets[socket].discover = true; + self.send(socket, reply(serde_json::json!({}))); + } + "Target.attachToTarget" => { + let wanted = params["targetId"].as_str().unwrap_or_default().to_string(); + if !self.pages.contains(&wanted) { + self.send( + socket, + serde_json::json!({ "id": id, "error": { "code": -32602, "message": "No target with given id found" } }), + ); + return; + } + let session = format!("S{socket}-{}", self.sockets[socket].sessions.len() + 1); + self.sockets[socket] + .sessions + .insert(session.clone(), wanted); + self.send(socket, reply(serde_json::json!({ "sessionId": session }))); + } + "Target.createTarget" => { + self.next_page += 1; + let page = format!("N{}", self.next_page); + self.pages.push(page.clone()); + for index in 0..self.sockets.len() { + if self.sockets[index].discover { + self.send( + index, + serde_json::json!({ "method": "Target.targetCreated", "params": { "targetInfo": Self::info(&page) } }), + ); + } + } + self.send(socket, reply(serde_json::json!({ "targetId": page }))); + } + "Page.enable" => { + let page = page.unwrap_or_default(); + self.sockets[socket].enabled.push(session.clone()); + match self.alerts.get(&page).cloned() { + Some(message) => { + if self.reannounce { + self.send( + socket, + frame( + &session, + serde_json::json!({ "method": "Page.javascriptDialogOpening", "params": Self::alert_event(&message) }), + ), + ); + } + self.sockets[socket].held.push((session, id)); + } + None => self.send(socket, reply(serde_json::json!({}))), + } + } + "Page.handleJavaScriptDialog" => { + let page = page.unwrap_or_default(); + if self.alerts.contains_key(&page) { + self.send(socket, reply(serde_json::json!({}))); + self.close_alert(&page); + } else { + self.send( + socket, + frame( + &session, + serde_json::json!({ "id": id, "error": { "code": -32000, "message": "No dialog is showing" } }), + ), + ); + } + } + "Runtime.evaluate" => { + let page = page.unwrap_or_default(); + if self.alerts.contains_key(&page) { + return; + } + if params["expression"] == "alert" { + self.send( + socket, + reply(serde_json::json!({ "result": { "type": "undefined" } })), + ); + self.open_alert(&page, "Saved?"); + } else { + self.send( + socket, + reply(serde_json::json!({ "result": { "type": "string", "value": page } })), + ); + } + } + _ => self.send(socket, reply(serde_json::json!({}))), + } + } + } + + /// Serve a [`Browser`] with `pages`. `/devtools/page/` is a page + /// socket; any other path is the browser, as a relay is. + async fn serve_browser(pages: &[&str], reannounce: bool) -> (u16, Arc>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let browser = Arc::new(Mutex::new(Browser { + pages: pages.iter().map(|page| page.to_string()).collect(), + reannounce, + ..Browser::default() + })); + let served = browser.clone(); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let browser = served.clone(); + tokio::spawn(async move { + let path = Arc::new(Mutex::new(String::new())); + let seen = path.clone(); + #[allow(clippy::result_large_err)] + let callback = + move |request: &tokio_tungstenite::tungstenite::handshake::server::Request, + response| { + *lock(&seen) = request.uri().path().to_string(); + Ok(response) + }; + let Ok(stream) = tokio_tungstenite::accept_hdr_async(socket, callback).await else { + return; + }; + let page = lock(&path) + .strip_prefix("/devtools/page/") + .map(str::to_string); + let (mut sink, mut stream) = stream.split(); + let (out, mut queued) = mpsc::unbounded_channel::>(); + let index = { + let mut browser = lock(&browser); + browser.sockets.push(FakeSocket { + out, + page, + sessions: HashMap::new(), + enabled: Vec::new(), + held: Vec::new(), + discover: false, + }); + browser.sockets.len() - 1 + }; + tokio::spawn(async move { + while let Some(Some(text)) = queued.recv().await { + if sink.send(Message::Text(text.into())).await.is_err() { + return; + } + } + let _ = sink.close().await; + }); + while let Some(Ok(Message::Text(text))) = stream.next().await { + if let Ok(request) = serde_json::from_str::(&text) { + lock(&browser).handle(index, request); + } + } + }); + } + }); + (port, browser) + } + + fn cloud_session(port: u16, session: &str) -> CdpTarget { + CdpTarget::Remote { + ws_url: format!("ws://127.0.0.1:{port}/devtools/browser/fake"), + bearer: "token".to_string(), + session_id: session.to_string(), + } + } + + async fn where_is(connection: &mut crate::cdp_target::CdpConnection, id: u64) -> Value { + connection + .call( + id, + "Runtime.evaluate", + serde_json::json!({ "expression": "where" }), + ) + .await + .unwrap()["result"]["value"] + .clone() + } + + async fn until(what: &str, done: impl Fn() -> bool) { + let waited = tokio::time::timeout(Duration::from_secs(3), async { + while !done() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + assert!(waited.is_ok(), "{what}"); + } + + fn forget(target: &CdpTarget) { + let key = target.pool_key(); + lock(&LINKS).remove(&key); + lock(&WATCHES).retain(|kept, _| !kept.starts_with(&key)); + } + + #[tokio::test] + async fn a_cloud_session_drives_one_pinned_page_and_watches_every_page_on_one_socket() { + let (port, browser) = serve_browser(&["A", "B"], false).await; + let target = cloud_session(port, "pinned"); + assert_eq!(watch(&target).await.unwrap(), PageState::Ready); + + // The browser lists its tabs in another order: the tools stay on the + // watched page, and attach to it without asking for the list. + lock(&browser).pages.reverse(); + let mut connection = target.connect().await.unwrap(); + assert_eq!(where_is(&mut connection, 1).await, "A"); + let socket = lock(&browser).sockets.len() - 1; + assert!(!lock(&browser) + .requests + .iter() + .any(|(on, request)| *on == socket && request["method"] == "Target.getTargets")); + + // An alert on it fails the next command of every connection at once. + connection + .call( + 2, + "Runtime.evaluate", + serde_json::json!({ "expression": "alert" }), + ) + .await + .unwrap(); + until("the alert is seen", || open_dialog(&target).is_some()).await; + let started = tokio::time::Instant::now(); + assert!(matches!( + where_is_result(&mut connection, 3).await, + Err(CdpError::DialogOpen(ref d)) if d.message == "Saved?" + )); + let mut other = target.connect().await.unwrap(); + assert!(matches!( + where_is_result(&mut other, 1).await, + Err(CdpError::DialogOpen(_)) + )); + assert!(started.elapsed() < Duration::from_millis(500)); + + // The page closes with its alert open: the tools move to the page left, + // which no dialog holds, and its dialogs are seen on the same socket. + lock(&browser).close_page("A"); + let page_a = target.page_key(Some("A")); + until("the closed page is no longer watched", || { + registered(&page_a).is_none() + }) + .await; + assert_eq!(watch(&target).await.unwrap(), PageState::Ready); + assert_eq!(remote_page(&target).as_deref(), Some("B")); + let mut moved = target.connect().await.unwrap(); + assert_eq!(where_is(&mut moved, 1).await, "B"); + moved + .call( + 2, + "Runtime.evaluate", + serde_json::json!({ "expression": "alert" }), + ) + .await + .unwrap(); + until("the second alert is seen", || { + open_dialog(&target).is_some() + }) + .await; + let answered = answer(&target, true, None).await.unwrap(); + assert_eq!(answered.dialog.message, "Saved?"); + assert_eq!(where_is(&mut moved, 3).await, "B"); + + let browser = lock(&browser); + let watching: std::collections::HashSet = browser + .requests + .iter() + .filter(|(_, request)| request["method"] == "Page.enable") + .map(|(on, _)| *on) + .collect(); + assert_eq!(watching.len(), 1, "one socket watches both pages"); + drop(browser); + forget(&target); + } + + async fn where_is_result( + connection: &mut crate::cdp_target::CdpConnection, + id: u64, + ) -> Result { + connection + .call( + id, + "Runtime.evaluate", + serde_json::json!({ "expression": "where" }), + ) + .await + } + + #[tokio::test] + async fn a_dialog_announced_again_to_a_new_watch_is_seen_at_once_and_answered() { + // A browser may tell a session that turns the Page domain on about a + // dialog that opened before it, and answer its Page.enable only after the + // dialog closes. + for remote in [false, true] { + let (port, browser) = serve_browser(&["A"], true).await; + lock(&browser).open_alert("A", "Opened before"); + let target = if remote { + cloud_session(port, "reannounced") + } else { + CdpTarget::Local { + ws_url: format!("ws://127.0.0.1:{port}/devtools/page/A"), + } + }; + let started = tokio::time::Instant::now(); + let state = watch(&target).await.unwrap(); + assert!( + matches!(state, PageState::Dialog(ref d) if d.message == "Opened before"), + "remote={remote} {state:?}" + ); + assert!( + started.elapsed() < Duration::from_secs(1), + "remote={remote}" + ); + + let mut connection = target.connect().await.unwrap(); + assert!(matches!( + where_is_result(&mut connection, 1).await, + Err(CdpError::DialogOpen(_)) + )); + let answered = answer(&target, false, None).await.unwrap(); + assert_eq!(answered.dialog.message, "Opened before"); + assert!(answered.next.is_none()); + // The Page.enable answer that comes with the close changes nothing. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(watch(&target).await.unwrap(), PageState::Ready); + assert_eq!(where_is(&mut connection, 2).await, "A"); + forget(&target); + } + } }