Spaces:
Running
Running
File size: 2,668 Bytes
2415446 0a54372 2415446 0a54372 2415446 0a54372 2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """Shared process helpers for installed client CLI launchers."""
import shutil
import subprocess
import sys
from collections.abc import Mapping
from urllib.error import HTTPError, URLError
from urllib.request import Request
from free_claude_code.cli.local_http import open_local_request
from free_claude_code.cli.process_registry import (
kill_pid_tree_best_effort,
register_pid,
unregister_pid,
)
PROXY_PREFLIGHT_PATH = "/health"
PROXY_PREFLIGHT_TIMEOUT_SECONDS = 1.5
def preflight_proxy(proxy_root_url: str) -> str | None:
"""Return an error message when the local proxy health check is unreachable."""
url = f"{proxy_root_url.rstrip('/')}{PROXY_PREFLIGHT_PATH}"
request = Request(url, method="GET")
try:
with open_local_request(
request, timeout=PROXY_PREFLIGHT_TIMEOUT_SECONDS
) as response:
status_code = response.getcode()
except HTTPError as exc:
return f"returned HTTP {exc.code}"
except URLError as exc:
return str(exc.reason)
except OSError as exc:
return str(exc)
if not 200 <= status_code < 300:
return f"returned HTTP {status_code}"
return None
def resolve_client_binary(
*,
binary_name: str,
display_name: str,
install_hint: str,
) -> str:
"""Resolve an installed client binary or exit with a user-facing hint."""
client_command = shutil.which(binary_name)
if client_command is None:
print(
f"Could not find {display_name} command: {binary_name}",
file=sys.stderr,
)
print(install_hint, file=sys.stderr)
raise SystemExit(127)
return client_command
def run_client_process(
*,
command: list[str],
env: Mapping[str, str],
binary_name: str,
display_name: str,
install_hint: str,
) -> None:
"""Run a client CLI command and mirror its exit code."""
process: subprocess.Popen[bytes] | None = None
try:
process = subprocess.Popen(command, env=dict(env))
if process.pid:
register_pid(process.pid)
return_code = process.wait()
except FileNotFoundError:
print(
f"Could not find {display_name} command: {binary_name}",
file=sys.stderr,
)
print(install_hint, file=sys.stderr)
raise SystemExit(127) from None
except KeyboardInterrupt:
if process is not None and process.pid:
kill_pid_tree_best_effort(process.pid)
process.wait()
raise
finally:
if process is not None and process.pid:
unregister_pid(process.pid)
raise SystemExit(return_code)
|