mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-07-03 14:05:26 +02:00
Ali Khokhar
6bee3104fe
## Problem The CLI package preserved a generic adapter layer and managed Codex parser path that did not match the supported customer workflows. Messaging runs Claude Code sessions, while Codex is supported through `fcc-codex` and extensions. ## Changes | Before | After | | --- | --- | | `fcc-claude` and `fcc-codex` shared generic adapter plumbing. | `fcc-claude` and `fcc-codex` use explicit launcher modules. | | Messaging depended on a generic CLI session abstraction. | Messaging depends on managed Claude Code sessions. | | Codex catalog generation lived as a top-level CLI helper. | Codex catalog generation lives under the Codex launcher owner. | | Tests asserted deleted internal adapter shapes. | Tests assert launcher, managed-Claude, and customer-surface behavior. |
94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
"""Shared process helpers for installed client CLI launchers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Mapping
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
from 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 urlopen(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)
|