Add desktop shell prompt bridge

README.md

@@ -41,6 +41,8 @@ python3 -m ampbrowser plan gemini://wownero.org/
 python3 -m ampbrowser open https://wownero.org/
 python3 -m ampbrowser open http://example.onion/ --launch
 python3 -m ampbrowser open http://example.onion/ --yes --launch
+python3 -m ampbrowser shell http://example.onion/
+python3 -m ampbrowser shell http://example.onion/ --yes
 python3 -m ampbrowser open http://example.b32.i2p/ --yes
 python3 -m ampbrowser transport start tor
 python3 -m ampbrowser transport status tor

ampbrowser/cli.py

@@ -1,15 +1,16 @@
 from __future__ import annotations
 
 import argparse
-from dataclasses import replace
 from pathlib import Path
 import shlex
 import sys
 
 from .config import load_config
+from .desktop_shell import launch_desktop_shell
 from .docsgen import generate_docs
 from .fixtures import check_fixture_manifest
-from .launch import execute_open, prepare_open
+from .launch import prepare_open
+from .open_runner import launch_open_plan
 from .plan import plan_url
 from .platforms import PLATFORM_CHOICES
 from .routing import route_url
@@ -51,6 +52,15 @@ def main(argv: list[str] | None = None) -> int:
         help="Create the isolated profile and launch the bundled browser when the route is ready.",
     )
 
+    shell_parser = subcommands.add_parser("shell", help="Launch the desktop shell flow.")
+    shell_parser.add_argument("url")
+    shell_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.")
+    shell_parser.add_argument(
+        "--yes",
+        action="store_true",
+        help="Approve first-use setup without showing a prompt.",
+    )
+
     subcommands.add_parser("inspect", help="Inspect local transport readiness.")
 
     transport_parser = subcommands.add_parser("transport", help="Manage AMPB-owned transports.")
@@ -90,6 +100,8 @@ def main(argv: list[str] | None = None) -> int:
                 config_path=args.config,
                 platform=args.platform,
             )
+        if args.command == "shell":
+            return _cmd_shell(args.url, approve=args.yes, config_path=args.config)
         if args.command == "inspect":
             return _cmd_inspect()
         if args.command == "transport":
@@ -154,26 +166,26 @@ def _cmd_open(
     config = load_config(Path.cwd(), config_path)
     open_plan = prepare_open(url, consent=consent, dry_run=dry_run, config=config, platform=platform)
     if launch:
-        if open_plan.status == "setup-approved":
-            transport_result = ensure_transport_ready(
-                open_plan.browse_plan.route.transport,
-                config=config,
-                root=Path.cwd(),
-                status=open_plan.browse_plan.status,
-            )
-            open_plan = replace(
-                open_plan,
-                dry_run=False,
-                status="ready" if transport_result.ready else transport_result.status,
-                message=transport_result.message,
-                transport_setup_status=transport_result.status,
-                transport_setup_provider=transport_result.provider,
-                transport_setup_owned=transport_result.owned,
-                transport_setup_pid=transport_result.pid,
-                transport_setup_endpoint=transport_result.endpoint,
-                transport_setup_message=transport_result.message,
-            )
-        open_plan = execute_open(open_plan, root=Path.cwd())
+        open_plan = launch_open_plan(open_plan, config=config, root=Path.cwd())
+    _print_open_plan("AMPBROWSER_OPEN", open_plan)
+    return 0
+
+
+def _cmd_shell(url: str, *, approve: bool, config_path: Path | None) -> int:
+    config = load_config(Path.cwd(), config_path)
+    result = launch_desktop_shell(url, config=config, root=Path.cwd(), auto_approve=approve)
+    _print_open_plan(
+        "AMPBROWSER_SHELL",
+        result.open_plan,
+        (
+            f"prompt_shown={str(result.prompt_shown).lower()}",
+            f"prompt_approved={str(result.prompt_approved).lower()}",
+        ),
+    )
+    return 0
+
+
+def _print_open_plan(prefix: str, open_plan, extra_fields: tuple[str, ...] = ()) -> None:
     browse_plan = open_plan.browse_plan
     setup_steps = "|".join(open_plan.setup_steps) if open_plan.setup_steps else "-"
     launch_spec = open_plan.launch_spec
@@ -181,8 +193,9 @@ def _cmd_open(
     runtime_path = launch_spec.runtime_path if launch_spec else "-"
     user_js_path = launch_spec.user_js_path if launch_spec else "-"
     message = _safe(open_plan.message)
+    extra = (" " + " ".join(extra_fields)) if extra_fields else ""
     print(
-        "AMPBROWSER_OPEN "
+        f"{prefix} "
         f"url={browse_plan.route.normalized} "
         f"transport={browse_plan.route.transport} "
         f"profile={browse_plan.route.profile} "
@@ -212,8 +225,8 @@ def _cmd_open(
         f"launch_command=\"{_safe(launch_command)}\" "
         f"setup_steps=\"{_safe(setup_steps)}\" "
         f"message=\"{message}\""
+        f"{extra}"
     )
-    return 0
 
 
 def _cmd_inspect() -> int:

ampbrowser/desktop_shell.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from dataclasses import replace
+from pathlib import Path
+import sys
+
+from .config import AppConfig
+from .launch import OpenPlan, prepare_open
+from .open_runner import launch_open_plan
+
+
+PromptDriver = Callable[[OpenPlan], bool]
+
+
+@dataclass(frozen=True)
+class DesktopShellResult:
+    open_plan: OpenPlan
+    prompt_shown: bool
+    prompt_approved: bool
+
+
+def launch_desktop_shell(
+    raw_url: str,
+    *,
+    config: AppConfig,
+    root: Path,
+    auto_approve: bool = False,
+    prompt_driver: PromptDriver | None = None,
+) -> DesktopShellResult:
+    open_plan = prepare_open(raw_url, consent=False, dry_run=False, config=config, platform="desktop")
+    prompt_shown = False
+    prompt_approved = False
+
+    if open_plan.status == "consent-required":
+        if auto_approve:
+            prompt_approved = True
+        else:
+            prompt_shown = True
+            prompt_approved = (prompt_driver or prompt_for_setup)(open_plan)
+        if not prompt_approved:
+            return DesktopShellResult(
+                replace(open_plan, dry_run=False, status="cancelled", message="first-use setup cancelled"),
+                prompt_shown,
+                False,
+            )
+        open_plan = prepare_open(raw_url, consent=True, dry_run=False, config=config, platform="desktop")
+
+    return DesktopShellResult(
+        launch_open_plan(open_plan, config=config, root=root),
+        prompt_shown,
+        prompt_approved,
+    )
+
+
+def prompt_for_setup(open_plan: OpenPlan) -> bool:
+    try:
+        return _tk_prompt(open_plan)
+    except Exception:  # noqa: BLE001
+        return _terminal_prompt(open_plan)
+
+
+def _tk_prompt(open_plan: OpenPlan) -> bool:
+    import tkinter as tk
+    from tkinter import ttk
+
+    result = {"approved": False}
+    window = tk.Tk()
+    window.title(open_plan.setup_prompt_title)
+    window.geometry("520x260")
+    window.minsize(420, 220)
+    window.resizable(False, False)
+
+    frame = ttk.Frame(window, padding=20)
+    frame.pack(fill="both", expand=True)
+
+    title = ttk.Label(frame, text=open_plan.setup_prompt_title, font=("TkDefaultFont", 15, "bold"))
+    title.pack(anchor="w")
+
+    body = ttk.Label(frame, text=open_plan.setup_prompt_body, wraplength=460, justify="left")
+    body.pack(anchor="w", fill="x", pady=(12, 18))
+
+    button_row = ttk.Frame(frame)
+    button_row.pack(anchor="e", fill="x")
+
+    def approve() -> None:
+        result["approved"] = True
+        window.quit()
+
+    def cancel() -> None:
+        result["approved"] = False
+        window.quit()
+
+    cancel_button = ttk.Button(button_row, text="Cancel", command=cancel)
+    cancel_button.pack(side="right")
+    approve_button = ttk.Button(button_row, text=open_plan.setup_prompt_approve_label, command=approve)
+    approve_button.pack(side="right", padx=(0, 8))
+    approve_button.focus_set()
+    window.bind("<Return>", lambda event: approve())
+    window.bind("<Escape>", lambda event: cancel())
+    window.protocol("WM_DELETE_WINDOW", cancel)
+    window.attributes("-topmost", True)
+    window.after(250, lambda: window.attributes("-topmost", False))
+    window.mainloop()
+    window.destroy()
+    return result["approved"]
+
+
+def _terminal_prompt(open_plan: OpenPlan) -> bool:
+    print(open_plan.setup_prompt_title, file=sys.stderr)
+    print(open_plan.setup_prompt_body, file=sys.stderr)
+    if not sys.stdin.isatty():
+        return False
+    answer = input(f"{open_plan.setup_prompt_approve_label}? [y/N] ")
+    return answer.strip().lower() in {"y", "yes"}

ampbrowser/open_runner.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+
+from .config import AppConfig
+from .launch import OpenPlan, execute_open
+from .transport_manager import ensure_transport_ready
+
+
+def launch_open_plan(open_plan: OpenPlan, *, config: AppConfig, root: Path) -> OpenPlan:
+    if open_plan.status == "setup-approved":
+        transport_result = ensure_transport_ready(
+            open_plan.browse_plan.route.transport,
+            config=config,
+            root=root,
+            status=open_plan.browse_plan.status,
+        )
+        open_plan = replace(
+            open_plan,
+            dry_run=False,
+            status="ready" if transport_result.ready else transport_result.status,
+            message=transport_result.message,
+            transport_setup_status=transport_result.status,
+            transport_setup_provider=transport_result.provider,
+            transport_setup_owned=transport_result.owned,
+            transport_setup_pid=transport_result.pid,
+            transport_setup_endpoint=transport_result.endpoint,
+            transport_setup_message=transport_result.message,
+        )
+    return execute_open(open_plan, root=root)

docs/desktop-vertical.md

@@ -30,3 +30,15 @@ The first transport target is Tor through an AMPB-owned Arti/Tor runtime. The de
 vertical is not complete until AMPB can start the runtime, verify the SOCKS endpoint, and
 launch the bundled browser profile without clearnet fallback. I2P and Reticulum use the
 same route, consent, state, and health-check model after Tor works.
+
+## Desktop Shell
+
+`ampbrowser shell <url>` is the first desktop shell bridge. It prepares the same open plan
+as `ampbrowser open`, renders first-use transport consent as a desktop dialog when a
+prompt is required, and then launches the bundled browser through the shared runner after
+approval. Use `--yes` for scripted verification without showing the dialog.
+
+```sh
+python3 -m ampbrowser shell http://example.onion/
+python3 -m ampbrowser shell http://example.onion/ --yes
+```

docs/open-flow.md

@@ -30,6 +30,10 @@ render those fields as the first-use popup. After approval, launch output includ
 `transport_setup_pid`, and `transport_setup_endpoint` so the shell can distinguish an
 adopted daemon from an AMPB-owned managed process.
 
+`ampbrowser shell <url>` is the desktop wrapper for this contract. It shows the prompt
+through a desktop dialog when possible, falls back to a terminal prompt when needed, and
+launches through the same transport runner after approval.
+
 ## Examples
 
 ```sh
@@ -41,4 +45,5 @@ python3 -m ampbrowser open http://example.b32.i2p/ --platform android
 python3 -m ampbrowser open https://wownero.org/ --launch
 python3 -m ampbrowser open http://example.onion/ --launch
 python3 -m ampbrowser open http://example.onion/ --yes --launch
+python3 -m ampbrowser shell http://example.onion/
 ```

tests/test_cli_open.py

@@ -8,6 +8,9 @@ import unittest
 from unittest.mock import patch
 
 from ampbrowser.cli import main
+from ampbrowser.config import AppConfig
+from ampbrowser.desktop_shell import DesktopShellResult
+from ampbrowser.launch import prepare_open
 from ampbrowser.transport_manager import ManagedTransportResult
 from ampbrowser.transports import TransportStatus
 
@@ -25,7 +28,7 @@ class CliOpenTest(unittest.TestCase):
         )
 
         with patch("ampbrowser.plan.inspect_transport", return_value=status):
-            with patch("ampbrowser.cli.ensure_transport_ready") as ensure_transport_ready:
+            with patch("ampbrowser.open_runner.ensure_transport_ready") as ensure_transport_ready:
                 with patch.object(sys, "stdout", new_callable=io.StringIO) as stdout:
                     code = main(["open", "http://example.onion/", "--launch"])
 
@@ -69,8 +72,8 @@ class CliOpenTest(unittest.TestCase):
             )
 
         with patch("ampbrowser.plan.inspect_transport", return_value=status):
-            with patch("ampbrowser.cli.ensure_transport_ready", return_value=transport_result) as ensure_transport_ready:
-                with patch("ampbrowser.cli.execute_open", side_effect=fake_execute) as execute_open:
+            with patch("ampbrowser.open_runner.ensure_transport_ready", return_value=transport_result) as ensure_transport_ready:
+                with patch("ampbrowser.open_runner.execute_open", side_effect=fake_execute) as execute_open:
                     with patch.object(sys, "stdout", new_callable=io.StringIO) as stdout:
                         code = main(["open", "http://example.onion/", "--yes", "--launch"])
 
@@ -86,6 +89,23 @@ class CliOpenTest(unittest.TestCase):
         self.assertIn("transport_setup_pid=1234", output)
         self.assertIn("transport_setup_endpoint=socks5://127.0.0.1:9050", output)
 
+    def test_shell_command_prints_desktop_shell_result(self) -> None:
+        config = AppConfig(runtime_path="/opt/ampb/firefox", transport_modes={})
+        open_plan = prepare_open("https://wownero.org/", config=config)
+        result = DesktopShellResult(open_plan=open_plan, prompt_shown=False, prompt_approved=False)
+
+        with patch("ampbrowser.cli.launch_desktop_shell", return_value=result) as shell:
+            with patch.object(sys, "stdout", new_callable=io.StringIO) as stdout:
+                code = main(["shell", "https://wownero.org/"])
+
+        self.assertEqual(0, code)
+        shell.assert_called_once()
+        output = stdout.getvalue()
+        self.assertIn("AMPBROWSER_SHELL", output)
+        self.assertIn("status=ready", output)
+        self.assertIn("prompt_shown=false", output)
+        self.assertIn("prompt_approved=false", output)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_desktop_shell.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from ampbrowser.config import AppConfig
+from ampbrowser.desktop_shell import launch_desktop_shell
+from ampbrowser.transports import TransportStatus
+
+
+class DesktopShellTest(unittest.TestCase):
+    def test_shell_cancel_does_not_launch(self) -> None:
+        status = TransportStatus(
+            transport="tor",
+            installed=True,
+            running=False,
+            endpoint="socks5://127.0.0.1:9050",
+            adoptable=False,
+            manage_supported=True,
+            note="Tor SOCKS proxy",
+        )
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch("ampbrowser.plan.inspect_transport", return_value=status):
+                with patch("ampbrowser.desktop_shell.launch_open_plan") as launch:
+                    result = launch_desktop_shell(
+                        "http://example.onion/",
+                        config=AppConfig(transport_modes={}),
+                        root=Path(tmp),
+                        prompt_driver=lambda open_plan: False,
+                    )
+
+        self.assertEqual("cancelled", result.open_plan.status)
+        self.assertTrue(result.prompt_shown)
+        self.assertFalse(result.prompt_approved)
+        launch.assert_not_called()
+
+    def test_shell_approval_launches_approved_plan(self) -> None:
+        status = TransportStatus(
+            transport="tor",
+            installed=True,
+            running=False,
+            endpoint="socks5://127.0.0.1:9050",
+            adoptable=False,
+            manage_supported=True,
+            note="Tor SOCKS proxy",
+        )
+
+        def fake_launch(open_plan, *, config: AppConfig, root: Path):
+            self.assertEqual("setup-approved", open_plan.status)
+            return replace(open_plan, status="launched", dry_run=False, browser_pid=1234)
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch("ampbrowser.plan.inspect_transport", return_value=status):
+                with patch("ampbrowser.desktop_shell.launch_open_plan", side_effect=fake_launch) as launch:
+                    result = launch_desktop_shell(
+                        "http://example.onion/",
+                        config=AppConfig(transport_modes={}),
+                        root=Path(tmp),
+                        prompt_driver=lambda open_plan: True,
+                    )
+
+        self.assertEqual("launched", result.open_plan.status)
+        self.assertEqual(1234, result.open_plan.browser_pid)
+        self.assertTrue(result.prompt_shown)
+        self.assertTrue(result.prompt_approved)
+        launch.assert_called_once()
+
+    def test_shell_auto_approve_skips_prompt_driver(self) -> None:
+        status = TransportStatus(
+            transport="tor",
+            installed=True,
+            running=False,
+            endpoint="socks5://127.0.0.1:9050",
+            adoptable=False,
+            manage_supported=True,
+            note="Tor SOCKS proxy",
+        )
+
+        def fake_launch(open_plan, *, config: AppConfig, root: Path):
+            return replace(open_plan, status="launched", dry_run=False, browser_pid=1234)
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch("ampbrowser.plan.inspect_transport", return_value=status):
+                with patch("ampbrowser.desktop_shell.launch_open_plan", side_effect=fake_launch):
+                    result = launch_desktop_shell(
+                        "http://example.onion/",
+                        config=AppConfig(transport_modes={}),
+                        root=Path(tmp),
+                        auto_approve=True,
+                        prompt_driver=lambda open_plan: False,
+                    )
+
+        self.assertEqual("launched", result.open_plan.status)
+        self.assertFalse(result.prompt_shown)
+        self.assertTrue(result.prompt_approved)
+
+
+if __name__ == "__main__":
+    unittest.main()