skills/deploy-handoff/handoff.py (30194 bytes)
1 """Open a web page and wait while the user does a step that only a human may do. 2 3 A tool calls this script when it reaches a step that an AI agent must not do: 4 the final deploy, publish, or billing click, a sign-in, an OAuth consent, or 5 the creation of a GitHub pull request. The script opens the page in the Brave 6 profile that brave_profile names in the user config file, or else in the default 7 browser. It shows a small dialog with the steps. It never clicks for the user. 8 It prints the answer of the user as one JSON object on stdout. 9 10 The command "terminal" shows the same dialog for a step at a prompt in a terminal, 11 for example the install question of a tool or a sign-in of a command line program. 12 It opens no page. The caller starts the program in a terminal that the user sees, 13 and stops at the prompt. The script never types the answer for the user. 14 15 The command "browser --fresh" starts a separate browser with a fresh profile for 16 the browser-driver agent of this plugin. Use it only if the user asks for it. 17 Playwright MCP controls that browser through a local port. 18 19 Exit codes: 0 done, started, or running, 1 error, 2 bad arguments, 3 not done, 20 4 timeout. 21 """ 22 23 import argparse 24 import contextlib 25 import ctypes 26 import http.client 27 import json 28 import os 29 import re 30 import shutil 31 import subprocess 32 import sys 33 import time 34 import tomllib 35 import urllib.request 36 import webbrowser 37 from collections.abc import Callable, Iterable 38 from dataclasses import dataclass 39 from pathlib import Path 40 from urllib.parse import quote, urlencode, urlsplit 41 42 # A Wayland desktop without X11 has no Tk. Then MenuDialog shows the steps in bemenu. 43 try: 44 import tkinter as tk 45 from tkinter import font as tkfont 46 from tkinter import ttk 47 48 HAVE_TK = True 49 except ImportError: 50 HAVE_TK = False 51 52 USER_CONFIG = ( 53 Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") 54 / "deploy-handoff" 55 / "config.toml" 56 ) 57 # The browser-driver agent uses a browser with a fresh profile of its own. Playwright MCP 58 # controls it through this remote debugging port. .mcp.json at the root of the plugin uses the 59 # same port. 60 DEBUG_PORT = 9333 61 BROWSER_PROFILE = USER_CONFIG.with_name("browser") 62 # The time in seconds that a new browser gets to open DEBUG_PORT. 63 BROWSER_START_SECONDS = 30 64 # The browser listens on 127.0.0.1. A proxy must not get these requests. 65 LOCAL_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({})) 66 # GitHub refuses very long request lines. This limit keeps a margin. 67 MAX_PR_URL_LENGTH = 8000 68 EXIT_CODES = {"done": 0, "started": 0, "running": 0, "error": 1, "not_done": 3, "timeout": 4} 69 # ASD-STE100 Simplified Technical English allows 20 words in one instruction. 70 MAX_STEP_WORDS = 20 71 PR_STEPS = ( 72 "Make sure that the branches, the title, and the description are correct.", 73 'Click "Create pull request".', 74 ) 75 HEADING = "Do these steps in your browser:" 76 TERMINAL_HEADING = "Do these steps in the terminal:" 77 FINISH = ( 78 'When you finish, click "Done".\n' 79 'If you cannot finish, write the reason in the note. Then click "Not done".' 80 ) 81 WARNING = ( 82 "Claude does not click for you. Before you sign in, make sure that the address bar " 83 "shows the site above. Do not type a password, key, or card number in this window." 84 ) 85 TERMINAL_WARNING = ( 86 "Claude does not type the answer for you. Do not type a password, key, or card number " 87 "in this window." 88 ) 89 90 # A plain DNS name in lowercase ASCII. The script refuses all other host forms, 91 # so the browser and this script always read the same host from a URL. 92 HOST_RE = re.compile(r"[a-z0-9-]+(?:\.[a-z0-9-]+)+") 93 # Printable ASCII without a space. Callers percent-encode all other characters. 94 URL_CHARS_RE = re.compile(r"[!-~]+") 95 REPO_RE = re.compile(r"[A-Za-z0-9-]+/[A-Za-z0-9._-]+") 96 GITHUB_REMOTE_RE = re.compile( 97 r"(?:https://(?:[^@/]+@)?github\.com/|git@github\.com:|ssh://git@github\.com/)" 98 r"(?P<repo>[A-Za-z0-9-]+/[A-Za-z0-9._-]+?)(?:\.git)?/?" 99 ) 100 101 102 class HandoffError(Exception): 103 """The script cannot start the handoff. The message tells the caller why.""" 104 105 106 @dataclass(frozen=True) 107 class Request: 108 """A page to open, or a terminal, and the steps that the user must do there.""" 109 110 title: str 111 # None for a step in a terminal. Then place names the terminal. 112 url: str | None 113 steps: tuple[str, ...] 114 # For a pull request: the repository as "owner/name", and the head branch. 115 pr: tuple[str, str] | None = None 116 # For a step in a terminal: where the prompt is, for example the name of a terminal tab. 117 place: str | None = None 118 119 def where(self) -> str: 120 """Return the line that tells the user where to do the steps.""" 121 if self.url is None: 122 return f"Where: {self.place}" 123 return f"Site: {urlsplit(self.url).hostname}" 124 125 126 def check_url(url: str) -> None: 127 """Raise HandoffError if url is not an https URL with a plain host name. 128 129 There is no list of allowed hosts: the user asks Claude to use only sites he trusts. 130 The dialog shows the host, so the user can compare it with the address bar. 131 """ 132 if "\\" in url or not URL_CHARS_RE.fullmatch(url): 133 raise HandoffError( 134 f"The URL must be printable ASCII with no spaces or backslashes: {url!r}" 135 ) 136 try: 137 parts = urlsplit(url) 138 except ValueError as exc: 139 raise HandoffError(f"The URL is not valid: {url}") from exc 140 host = parts.hostname or "" 141 # The netloc must be the host only: no user name, no password, and no port. 142 if parts.scheme != "https" or parts.netloc.lower() != host or not HOST_RE.fullmatch(host): 143 raise HandoffError(f"The URL must use https and a plain host name: {url}") 144 145 146 def check_text(title: str, steps: Iterable[str]) -> None: 147 """Raise HandoffError if the title or a step is not short Simplified Technical English.""" 148 named = [("The title", title), *((f"Step {n}", step) for n, step in enumerate(steps, 1))] 149 for name, text in named: 150 words = len(text.split()) 151 if not 0 < words <= MAX_STEP_WORDS or ";" in text: 152 raise HandoffError( 153 f"{name} is not short Simplified Technical English ({words} words). " 154 f"Give one action in 1 to {MAX_STEP_WORDS} words, with no semicolons." 155 ) 156 157 158 def check_place(place: str) -> None: 159 """Raise HandoffError if place does not name a terminal in a few printable words.""" 160 words = len(place.split()) 161 if not 0 < words <= MAX_STEP_WORDS or not place.isprintable(): 162 raise HandoffError( 163 f"The place is not a short name of a terminal ({words} words). " 164 f"Give 1 to {MAX_STEP_WORDS} printable words, for example the name of the tab." 165 ) 166 167 168 def read_config(path: Path) -> dict[str, object]: 169 """Return the data of the TOML file at path.""" 170 try: 171 return tomllib.loads(path.read_text(encoding="utf-8")) 172 except (OSError, tomllib.TOMLDecodeError) as exc: 173 raise HandoffError(f"Cannot read {path}: {exc}") from exc 174 175 176 def brave_profile() -> str | None: 177 """Return the name of the Brave profile in the user file, or None if the file names none.""" 178 if not USER_CONFIG.is_file(): 179 return None 180 name = read_config(USER_CONFIG).get("brave_profile") 181 if name is None: 182 return None 183 if not isinstance(name, str) or not name.strip(): 184 raise HandoffError(f"{USER_CONFIG}: brave_profile must be the name of a Brave profile.") 185 return name.strip() 186 187 188 def brave_data_dir() -> Path: 189 """Return the folder in which Brave keeps its profiles.""" 190 if sys.platform == "win32": 191 local = Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local") 192 return local / "BraveSoftware" / "Brave-Browser" / "User Data" 193 if sys.platform == "darwin": 194 return Path.home() / "Library" / "Application Support" / "BraveSoftware" / "Brave-Browser" 195 config = Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") 196 return config / "BraveSoftware" / "Brave-Browser" 197 198 199 def find_profile(local_state: Path, name: str) -> str: 200 """Return the folder of the Brave profile with the display name name. Ignore the case.""" 201 try: 202 cache = json.loads(local_state.read_text(encoding="utf-8"))["profile"]["info_cache"] 203 folders = [ 204 str(folder) 205 for folder, info in cache.items() 206 if str(info.get("name", "")).casefold() == name.casefold() 207 ] 208 except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc: 209 raise HandoffError(f"Cannot read the Brave profiles in {local_state}: {exc!r}") from exc 210 if len(folders) != 1: 211 raise HandoffError( 212 f"Found {len(folders)} Brave profiles with the name {name!r} in {local_state}. " 213 f"Set brave_profile in {USER_CONFIG} to the name of one profile." 214 ) 215 return folders[0] 216 217 218 def open_in_profile(url: str, profile: str) -> None: 219 """Open url in a new tab of the Brave profile with the display name profile.""" 220 exe = next((path for path in brave_candidates() if path.is_file()), None) 221 if exe is None: 222 raise HandoffError(f"Found no Brave. {USER_CONFIG} sets brave_profile to {profile!r}.") 223 folder = find_profile(brave_data_dir() / "Local State", profile) 224 # If Brave runs already, this command gives the URL to that Brave and ends. 225 start_browser([str(exe), f"--profile-directory={folder}", url]) 226 227 228 def open_page(url: str) -> None: 229 """Open url in a new tab. BROWSER can name a different default browser. 230 231 Use the Brave profile of the user file if it names one. Then BROWSER has no effect. 232 """ 233 profile = brave_profile() 234 if profile is not None: 235 open_in_profile(url, profile) 236 elif not webbrowser.open(url, new=2): 237 raise HandoffError("No browser opened the page. Set the BROWSER environment variable.") 238 239 240 def browser_candidates() -> list[Path]: 241 """Return the usual paths of Brave and Chrome on this system. Brave comes first.""" 242 if sys.platform == "win32": 243 roots = [ 244 os.environ.get(name) for name in ("PROGRAMFILES", "PROGRAMFILES(X86)", "LOCALAPPDATA") 245 ] 246 programs = ( 247 r"BraveSoftware\Brave-Browser\Application\brave.exe", 248 r"Google\Chrome\Application\chrome.exe", 249 ) 250 return [Path(root, program) for program in programs for root in roots if root] 251 if sys.platform == "darwin": 252 return [ 253 Path("/Applications/Brave Browser.app/Contents/MacOS/Brave Browser"), 254 Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"), 255 ] 256 names = ("brave-browser", "brave", "google-chrome", "chromium", "chromium-browser") 257 return [Path(path) for name in names if (path := shutil.which(name))] 258 259 260 def brave_candidates() -> list[Path]: 261 """Return the usual paths of Brave on this system.""" 262 return [path for path in browser_candidates() if "brave" in path.name.lower()] 263 264 265 def find_browser(candidates: Iterable[Path]) -> Path: 266 """Return the first candidate that is a file.""" 267 for path in candidates: 268 if path.is_file(): 269 return path 270 raise HandoffError("Found no Brave or Chrome. Give --exe PATH.") 271 272 273 def browser_command(exe: Path, profile: Path, port: int) -> list[str]: 274 """Return the command that starts the browser of the browser-driver agent.""" 275 return [ 276 str(exe), 277 f"--user-data-dir={profile}", 278 f"--remote-debugging-port={port}", 279 # Start with no window. With this flag and a remote debugging port, Chromium keeps 280 # running after its last window closes. Without it, the browser stops when a person or 281 # a program closes its window, and the agent cannot connect to the port. 282 "--no-startup-window", 283 "--no-first-run", 284 "--no-default-browser-check", 285 # Keep the pages active when other windows cover the window of this browser. 286 "--disable-backgrounding-occluded-windows", 287 "--disable-renderer-backgrounding", 288 "--disable-background-timer-throttling", 289 ] 290 291 292 def debug_version(port: int) -> dict[str, object] | None: 293 """Return the version data of the browser that listens on port, or None if none answers.""" 294 try: 295 with LOCAL_OPENER.open(f"http://127.0.0.1:{port}/json/version", timeout=2) as response: 296 data = json.load(response) 297 except (OSError, ValueError, http.client.HTTPException): 298 return None 299 return data if isinstance(data, dict) else None 300 301 302 def start_browser(command: list[str]) -> None: 303 """Start the browser in a new process group, so that it runs after this script ends.""" 304 devnull = subprocess.DEVNULL 305 try: 306 if sys.platform == "win32": 307 flags = subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP 308 subprocess.Popen( 309 command, stdin=devnull, stdout=devnull, stderr=devnull, creationflags=flags 310 ) 311 else: 312 subprocess.Popen( 313 command, stdin=devnull, stdout=devnull, stderr=devnull, start_new_session=True 314 ) 315 except OSError as exc: 316 raise HandoffError(f"Cannot start {command[0]}: {exc}") from exc 317 318 319 def wait_for_browser(port: int, seconds: float) -> dict[str, object]: 320 """Wait until a browser answers on port. Return its version data.""" 321 deadline = time.monotonic() + seconds 322 while (version := debug_version(port)) is None: 323 if time.monotonic() > deadline: 324 raise HandoffError( 325 f"The browser did not open port {port} in {seconds:g} seconds. If the browser " 326 "of the agent runs without this port, close it. Then run this command again." 327 ) 328 time.sleep(0.5) 329 return version 330 331 332 def run_browser(exe: Path | None) -> dict[str, str | None]: 333 """Start the browser of the browser-driver agent, or find it running.""" 334 status = "running" 335 version = debug_version(DEBUG_PORT) 336 if version is None: 337 status = "started" 338 BROWSER_PROFILE.mkdir(parents=True, exist_ok=True) 339 exe = exe or find_browser(browser_candidates()) 340 start_browser(browser_command(exe, BROWSER_PROFILE, DEBUG_PORT)) 341 version = wait_for_browser(DEBUG_PORT, BROWSER_START_SECONDS) 342 endpoint = f"http://127.0.0.1:{DEBUG_PORT}" 343 return {"status": status, "browser": str(version.get("Browser")), "endpoint": endpoint} 344 345 346 def github_repo(remote_url: str) -> str: 347 """Return "owner/name" for the URL of a github.com remote.""" 348 match = GITHUB_REMOTE_RE.fullmatch(remote_url.strip()) 349 if match is None: 350 raise HandoffError(f"{remote_url} is not a github.com remote. Give --repo OWNER/NAME.") 351 return match["repo"] 352 353 354 def fork_owner(remote_url: str, repo: str) -> str | None: 355 """Return the owner of a GitHub remote that is not repo, for example a fork of repo, or None.""" 356 match = GITHUB_REMOTE_RE.fullmatch(remote_url.strip()) 357 if match is None or match["repo"].lower() == repo.lower(): 358 return None 359 return match["repo"].split("/")[0] 360 361 362 def compare_url(repo: str, base: str | None, head: str, title: str, body: str) -> str: 363 """Return the GitHub page that shows a new pull request form with title and body filled in. 364 365 head is BRANCH, or OWNER:BRANCH for a branch in a fork of repo. 366 """ 367 # Without a base, GitHub compares head with the default branch. A branch name has no ":". 368 head_ref = quote(head, safe="/:") 369 refs = head_ref if base is None else f"{quote(base)}...{head_ref}" 370 query = urlencode({"expand": "1", "title": title, "body": body}, quote_via=quote) 371 url = f"https://github.com/{repo}/compare/{refs}?{query}" 372 if len(url) > MAX_PR_URL_LENGTH: 373 raise HandoffError( 374 f"The pull request URL has {len(url)} characters. The limit is {MAX_PR_URL_LENGTH}. " 375 "Make the body shorter." 376 ) 377 return url 378 379 380 def git(repo_dir: Path, *args: str) -> str: 381 """Run git in repo_dir and return its output. Raise HandoffError if git fails.""" 382 command = ["git", "-C", str(repo_dir), *args] 383 try: 384 result = subprocess.run( 385 command, capture_output=True, encoding="utf-8", errors="replace", timeout=60 386 ) 387 except (OSError, subprocess.TimeoutExpired) as exc: 388 raise HandoffError(f"{' '.join(command)} failed: {exc}") from exc 389 if result.returncode != 0: 390 raise HandoffError(f"{' '.join(command)} failed: {result.stderr.strip()}") 391 return result.stdout.strip() 392 393 394 def current_branch(repo_dir: Path) -> str: 395 """Return the branch that HEAD is on.""" 396 branch = git(repo_dir, "branch", "--show-current") 397 if not branch: 398 raise HandoffError("HEAD is not on a branch. Give --head BRANCH.") 399 return branch 400 401 402 def check_pushed(repo_dir: Path, remote: str, branch: str) -> None: 403 """Raise HandoffError if remote does not have the commit of the local branch.""" 404 ref = f"refs/heads/{branch}" 405 local = git(repo_dir, "rev-parse", "--verify", ref) 406 lines = git(repo_dir, "ls-remote", remote, ref).splitlines() 407 remote_commits = [line.split()[0] for line in lines if line.split()[1:] == [ref]] 408 if remote_commits != [local]: 409 raise HandoffError( 410 f"{remote} does not have the local commit of {branch}. Push {branch} first." 411 ) 412 413 414 def find_open_pr(repo: str, head: str) -> str | None: 415 """Return the URL of the open pull request from head, or None if gh cannot find one. 416 417 head is BRANCH, or OWNER:BRANCH for a branch in a fork of repo. 418 """ 419 owner, _, branch = head.rpartition(":") 420 command = ["gh", "pr", "list", "--repo", repo, "--head", branch, "--state", "open"] 421 if owner: 422 # gh filters by the branch name only. The jq filter also checks the owner of the branch. 423 login = f'select(.headRepositoryOwner.login | ascii_downcase == "{owner.lower()}")' 424 command += ["--json", "url,headRepositoryOwner", "--jq", f"[.[] | {login}][0].url // empty"] 425 else: 426 command += ["--json", "url", "--jq", ".[0].url // empty"] 427 try: 428 result = subprocess.run(command, capture_output=True, encoding="utf-8", timeout=60) 429 except (OSError, subprocess.TimeoutExpired): 430 return None 431 url = result.stdout.strip() 432 return url if result.returncode == 0 and url else None 433 434 435 def pr_request(args: argparse.Namespace) -> Request: 436 """Make the request that opens the GitHub form for a new pull request.""" 437 repo_dir: Path = args.repo_dir 438 remote_url = git(repo_dir, "remote", "get-url", args.remote) 439 repo: str = args.repo or github_repo(remote_url) 440 if not REPO_RE.fullmatch(repo): 441 raise HandoffError(f"{repo!r} is not a repository name of the form OWNER/NAME.") 442 head: str = args.head or current_branch(repo_dir) 443 check_pushed(repo_dir, args.remote, head) 444 # A pull request to another repository, for example from a fork to its upstream 445 # repository, names the branch as OWNER:BRANCH. 446 owner = fork_owner(remote_url, repo) 447 head_ref = head if owner is None else f"{owner}:{head}" 448 try: 449 body: str = args.body_file.read_text(encoding="utf-8") if args.body_file else args.body 450 except OSError as exc: 451 raise HandoffError(f"Cannot read {args.body_file}: {exc}") from exc 452 url = compare_url(repo, args.base, head_ref, args.title, body) 453 return Request(f"Create pull request: {args.title}", url, PR_STEPS, pr=(repo, head_ref)) 454 455 456 class Dialog: 457 """A small window on top of the browser. It shows the steps and records the answer.""" 458 459 def __init__(self, request: Request, timeout_minutes: float, reopen: Callable[[], None]): 460 self.status = "not_done" 461 self.note_text = "" 462 if sys.platform == "win32": 463 # Draw sharp text on high-DPI screens, as IDLE does. The call fails if the 464 # process has already set the DPI awareness. That is not a problem. 465 with contextlib.suppress(OSError): 466 ctypes.OleDLL("shcore").SetProcessDpiAwareness(1) 467 self.root = tk.Tk() 468 scale = self.root.winfo_fpixels("1i") / 96 469 wrap = round(420 * scale) 470 self.root.title(f"Claude handoff: {request.title}") 471 self.root.attributes("-topmost", True) 472 self.root.resizable(False, False) 473 # Put the window in the bottom-left corner, above the taskbar. Deploy pages 474 # usually put the final button on the right side. 475 self.root.geometry(f"+{round(24 * scale)}-{round(80 * scale)}") 476 self.root.protocol("WM_DELETE_WINDOW", lambda: self.finish("not_done")) 477 478 steps = "\n".join(f"{number}. {step}" for number, step in enumerate(request.steps, 1)) 479 # Tk deletes a font when its Python object goes away, so the dialog keeps it. 480 self.bold = tkfont.nametofont("TkDefaultFont").copy() 481 self.bold.configure(weight="bold") 482 frame = ttk.Frame(self.root, padding=round(12 * scale)) 483 frame.grid() 484 # The user must see first what to do, and then where to do it. 485 ttk.Label(frame, text=request.title, font=self.bold, wraplength=wrap).grid(sticky="w") 486 heading = HEADING if request.url is not None else TERMINAL_HEADING 487 ttk.Label(frame, text=heading, font=self.bold).grid(sticky="w", pady=(8, 0)) 488 ttk.Label(frame, text=steps, wraplength=wrap, justify="left").grid(sticky="w") 489 where = ttk.Label(frame, text=request.where(), font=self.bold, wraplength=wrap) 490 where.grid(sticky="w", pady=(8, 0)) 491 warning = TERMINAL_WARNING 492 if request.url is not None: 493 # A step on a page: show the address, and open the page again on a click. 494 parts = urlsplit(request.url) 495 shown = f"https://{parts.netloc}{parts.path}" 496 if len(shown) > 80: 497 shown = shown[:77] + "..." 498 link = ttk.Label(frame, text=shown, foreground="blue", cursor="hand2", wraplength=wrap) 499 link.grid(sticky="w") 500 link.bind("<Button-1>", lambda _event: reopen()) 501 warning = WARNING 502 for text, color in ((FINISH, ""), (warning, "#b00020")): 503 label = ttk.Label(frame, text=text, foreground=color, wraplength=wrap, justify="left") 504 label.grid(sticky="w", pady=(8, 0)) 505 ttk.Label(frame, text="Note for Claude (optional):").grid(sticky="w", pady=(8, 0)) 506 self.note = ttk.Entry(frame) 507 self.note.grid(sticky="ew") 508 buttons = ttk.Frame(frame) 509 buttons.grid(sticky="e", pady=(12, 0)) 510 self.done_button = ttk.Button(buttons, text="Done", command=lambda: self.finish("done")) 511 self.done_button.grid(row=0, column=0, padx=(0, 8)) 512 self.not_done_button = ttk.Button( 513 buttons, text="Not done", command=lambda: self.finish("not_done") 514 ) 515 self.not_done_button.grid(row=0, column=1) 516 self.timer = self.root.after(round(timeout_minutes * 60_000), self.finish, "timeout") 517 518 def finish(self, status: str) -> None: 519 """Record the answer and close the window.""" 520 self.root.after_cancel(self.timer) 521 self.status = status 522 self.note_text = self.note.get().strip() 523 self.root.destroy() 524 525 def run(self) -> tuple[str, str]: 526 """Show the window until the user answers or the time ends.""" 527 self.root.mainloop() 528 return self.status, self.note_text 529 530 531 class MenuDialog: 532 """The steps in a bemenu list, for a Wayland desktop that has no Tk.""" 533 534 DONE = "Done" 535 NOT_DONE = "Not done" 536 537 def __init__(self, request: Request, timeout_minutes: float, reopen: Callable[[], None]): 538 if shutil.which("bemenu") is None: 539 raise HandoffError("Cannot show the steps: install Python Tk or bemenu.") 540 # Without a session, bemenu exits as if the user pressed Escape. 541 if not os.environ.get("WAYLAND_DISPLAY"): 542 raise HandoffError("Cannot show the steps: WAYLAND_DISPLAY is not set.") 543 self.title = request.title 544 self.deadline = time.monotonic() + timeout_minutes * 60 545 steps = [f"{number}. {step}" for number, step in enumerate(request.steps, 1)] 546 self.items = [ 547 *steps, 548 request.where(), 549 "If you cannot finish, type the reason and press Enter.", 550 self.DONE, 551 self.NOT_DONE, 552 ] 553 554 def run(self) -> tuple[str, str]: 555 """Show the list until the user selects an answer or the time ends.""" 556 while (remaining := self.deadline - time.monotonic()) > 0: 557 command = ["bemenu", "--list", str(len(self.items)), "--prompt", self.title] 558 try: 559 answer = subprocess.run( 560 command, 561 input="\n".join(self.items), 562 capture_output=True, 563 text=True, 564 timeout=remaining, 565 ) 566 except subprocess.TimeoutExpired: 567 break 568 choice = answer.stdout.strip() 569 # Escape makes bemenu exit with code 1. 570 if answer.returncode != 0 or choice == self.NOT_DONE: 571 return "not_done", "" 572 if choice == self.DONE: 573 return "done", "" 574 if choice not in self.items: 575 return "not_done", choice 576 # The user selected a step or the site. Show the list again. 577 return "timeout", "" 578 579 580 def parse_args(argv: list[str] | None) -> argparse.Namespace: 581 """Read the command line.""" 582 common = argparse.ArgumentParser(add_help=False) 583 common.add_argument( 584 "--timeout", 585 type=float, 586 default=30, 587 metavar="MINUTES", 588 help="the time to wait for the user (default: 30)", 589 ) 590 parser = argparse.ArgumentParser( 591 description="Open a web page and wait while the user does the final step." 592 ) 593 commands = parser.add_subparsers(dest="command", required=True) 594 page = commands.add_parser("open", parents=[common], help="open a page and show the steps") 595 page.add_argument("--url", required=True, help="the https URL of the page") 596 page.add_argument("--title", required=True, help="the action, for example: Publish 1.2") 597 page.add_argument( 598 "--step", action="append", required=True, help="one step for the user (give it again)" 599 ) 600 page.add_argument( 601 "--no-open", action="store_true", help="show only the dialog: the page is already open" 602 ) 603 terminal = commands.add_parser( 604 "terminal", parents=[common], help="show the steps for a prompt in a terminal" 605 ) 606 terminal.add_argument("--title", required=True, help="the action, for example: Install 1.2") 607 terminal.add_argument( 608 "--where", required=True, help="where the prompt is, for example the name of the tab" 609 ) 610 terminal.add_argument( 611 "--step", action="append", required=True, help="one step for the user (give it again)" 612 ) 613 pr = commands.add_parser("pr", parents=[common], help="open the GitHub pull request form") 614 pr.add_argument("--title", required=True, help="the title of the pull request") 615 body = pr.add_mutually_exclusive_group() 616 body.add_argument("--body", default="", help="the description of the pull request") 617 body.add_argument("--body-file", type=Path, help="a file with the description") 618 pr.add_argument("--base", help="the branch to merge into (default: the default branch)") 619 pr.add_argument("--head", help="the branch to merge (default: the current branch)") 620 pr.add_argument("--repo", help="OWNER/NAME on GitHub (default: from the remote URL)") 621 pr.add_argument("--remote", default="origin", help="the remote with the branch") 622 pr.add_argument("--repo-dir", type=Path, default=Path(), help="the local repository") 623 browser = commands.add_parser( 624 "browser", help="start a browser with a fresh profile for the browser-driver agent" 625 ) 626 browser.add_argument( 627 "--fresh", action="store_true", help="confirm that the user asked for a fresh profile" 628 ) 629 browser.add_argument("--exe", type=Path, help="the browser program (default: Brave or Chrome)") 630 return parser.parse_args(argv) 631 632 633 def report(result: dict[str, str | None]) -> int: 634 """Print result as JSON. Return the exit code for its status.""" 635 print(json.dumps(result)) 636 return EXIT_CODES[str(result["status"])] 637 638 639 def main(argv: list[str] | None = None) -> int: 640 """Do the handoff. Return the exit code.""" 641 args = parse_args(argv) 642 try: 643 if args.command == "browser": 644 if not args.fresh: 645 raise HandoffError( 646 "This command starts a browser with a fresh profile and no sign-in. Start it " 647 "only if the user asked for a fresh browser. Then give --fresh. Otherwise " 648 "use the Brave profile of the user." 649 ) 650 return report(run_browser(args.exe)) 651 if args.command == "terminal": 652 # A step at a prompt in a terminal has no page: no host to check, nothing to open. 653 check_text(args.title, args.step) 654 check_place(args.where) 655 request = Request(args.title, None, tuple(args.step), place=args.where) 656 else: 657 if args.command == "pr": 658 request = pr_request(args) 659 else: 660 check_text(args.title, args.step) 661 request = Request(args.title, args.url, tuple(args.step)) 662 check_url(str(request.url)) 663 # With --no-open, the caller drove a browser to the page already. 664 if args.command == "pr" or not args.no_open: 665 open_page(str(request.url)) 666 url = request.url 667 show = Dialog if HAVE_TK else MenuDialog 668 dialog = show(request, args.timeout, lambda: open_page(url) if url else None) 669 except HandoffError as exc: 670 return report({"status": "error", "error": str(exc)}) 671 status, note = dialog.run() 672 result: dict[str, str | None] = {"status": status, "note": note} 673 if request.pr is not None and status == "done": 674 result["pr_url"] = find_open_pr(*request.pr) 675 return report(result) 676 677 678 if __name__ == "__main__": 679 sys.exit(main())