diff --git a/src/together/lib/cli/__init__.py b/src/together/lib/cli/__init__.py index bc4193d2c..d5d363343 100644 --- a/src/together/lib/cli/__init__.py +++ b/src/together/lib/cli/__init__.py @@ -1,6 +1,5 @@ from __future__ import annotations -import os import sys import inspect from typing import Optional, Annotated, get_args, get_origin @@ -14,13 +13,19 @@ from together.lib.utils import log_debug from together._exceptions import APIError from together._utils._json import openapi_dumps -from together._utils._logs import setup_logging from together.lib.cli._track_cli import ( CliTrackingEvents, track_cli, flush_pending_events, format_cli_error_for_telemetry, ) +from together.lib.cli.utils._debug import ( + log_debug_note, + log_debug_session, + teardown_cli_debug, + setup_cli_debug_logging, + install_http_debug_hooks, +) from together.lib.cli.utils.config import CLIConfig from together.lib.cli.utils._prompt import PromptParameter from together.lib.cli.utils._console import console @@ -134,6 +139,12 @@ def _propagate_global_param_group(target_app: App) -> None: _propagate_global_param_group(sub) +# Stainless requires a non-empty key to construct a client. When the user has +# none, we substitute this placeholder and block real requests. Do not treat the +# placeholder as a real key in --debug session output. +_PLACEHOLDER_API_KEY = "0" * 40 + + def _create_client( api_key: Optional[str], base_url: Optional[str], @@ -141,7 +152,9 @@ def _create_client( max_retries: Optional[int], project_id: Optional[str], require_api_key: bool = True, -) -> AsyncTogether: + debug: bool = False, +) -> tuple[AsyncTogether, bool]: + missing_api_key = False try: client = AsyncTogether( api_key=api_key, @@ -152,22 +165,14 @@ def _create_client( ) except Exception as e: if "api_key" in str(e): + missing_api_key = True client = AsyncTogether( - api_key="0000000000000000000000000000000000000000", + api_key=_PLACEHOLDER_API_KEY, base_url=base_url, timeout=timeout, max_retries=max_retries if max_retries is not None else 0, project_id=project_id, ) - - def block_requests_for_api_key(_: httpx.Request) -> None: - console.print( - "[red]x[/red] api key missing.\n\nThe api key must be set either by passing --api-key to the command or by setting the TOGETHER_API_KEY environment variable", - ) - console.print("You can find your api key at https://api.together.ai/settings/api-keys") - sys.exit(1) - - client._client.event_hooks["request"].append(block_requests_for_api_key) else: raise e @@ -181,6 +186,21 @@ async def track_request(request: httpx.Request) -> None: except Exception as e: log_debug("Error tracking api request", error=e) + if debug: + install_http_debug_hooks(client._client) + + if missing_api_key: + # After debug hooks so `--debug` still emits `→ GET` before we exit, + # but before analytics so a request that is never sent is not tracked. + async def block_requests_for_api_key(_: httpx.Request) -> None: + console.print( + "[red]x[/red] api key missing.\n\nThe api key must be set either by passing --api-key to the command or by setting the TOGETHER_API_KEY environment variable", + ) + console.print("You can find your api key at https://api.together.ai/settings/api-keys") + sys.exit(1) + + client._client.event_hooks["request"].append(block_requests_for_api_key) + client._client.event_hooks["request"].append(track_request) # Out-of-band-auth commands (e.g. `beta clusters ssh`) make no Together API @@ -193,7 +213,7 @@ async def track_request(request: httpx.Request) -> None: console.print("You can find your api key at https://api.together.ai/settings/api-keys") sys.exit(1) - return client + return client, missing_api_key global_options = Group( @@ -208,7 +228,14 @@ async def launcher( base_url: Annotated[Optional[str], Parameter(show=False)] = None, timeout: Annotated[Optional[int], Parameter(show=False)] = None, max_retries: Annotated[Optional[int], Parameter(show=False)] = None, - debug: Annotated[Optional[bool], Parameter(show=False)] = False, + debug: Annotated[ + Optional[bool], + Parameter( + group=global_options, + negative=(), + help="Print HTTP request/response details to stderr", + ), + ] = False, non_interactive: Annotated[ Optional[bool], Parameter(group=global_options, negative=(), help="Disable interactive prompts") ] = False, @@ -231,9 +258,37 @@ async def launcher( ] = False, ) -> None: if debug: - os.environ.setdefault("TOGETHER_LOG", "debug") - setup_logging() + setup_cli_debug_logging() + try: + await _run_launcher( + tokens, + api_key=api_key, + base_url=base_url, + timeout=timeout, + max_retries=max_retries, + debug=debug, + non_interactive=non_interactive, + project_id=project_id, + output_json=output_json, + ) + finally: + if debug: + teardown_cli_debug() + + +async def _run_launcher( + tokens: tuple[str, ...], + *, + api_key: Optional[str], + base_url: Optional[str], + timeout: Optional[int], + max_retries: Optional[int], + debug: Optional[bool], + non_interactive: Optional[bool], + project_id: Optional[str], + output_json: Optional[bool], +) -> None: (parsed_command, explicit_args, is_beta_command, remaining) = preparse_tokens(app, [*tokens]) # Some commands authenticate out-of-band (OIDC / step-ca signed certificates) @@ -245,13 +300,34 @@ async def launcher( # they stay keyless. no_auth_command = is_beta_command and parsed_command in _NO_AUTH_COMMANDS - client = _create_client(api_key, base_url, timeout, max_retries, project_id, require_api_key=not no_auth_command) + client, missing_api_key = _create_client( + api_key, + base_url, + timeout, + max_retries, + project_id, + require_api_key=not no_auth_command, + debug=bool(debug), + ) + + if debug: + log_debug_session( + command=parsed_command, + is_beta_command=is_beta_command, + base_url=str(client.base_url), + project_id=client.project_id, + api_key=None if missing_api_key else (client.api_key or None), + timeout=client.timeout, + max_retries=client.max_retries, + ) # Skip the project-resolution whoami() for out-of-band-auth commands: it is a # Together API call and would reintroduce the API-key dependency for keyless # commands like `beta clusters ssh`. if not no_auth_command and client.project_id is None: client.project_id = await _resolve_project_id(client) + if debug and client.project_id: + log_debug_note(f"resolved project {client.project_id}") is_interactive = sys.stdin.isatty() and sys.stdout.isatty() and sys.stderr.isatty() and not _is_agent_or_ci() non_interactive_mode = non_interactive or output_json or not is_interactive diff --git a/src/together/lib/cli/api/beta/models/upload.py b/src/together/lib/cli/api/beta/models/upload.py index 8abdf023a..6339302b2 100644 --- a/src/together/lib/cli/api/beta/models/upload.py +++ b/src/together/lib/cli/api/beta/models/upload.py @@ -14,6 +14,7 @@ from together._utils import path_template from together.lib.cli.utils.config import CLIConfig, CLIConfigParameter from together.lib.cli.utils._console import console +from together.lib.cli.components.loader import loading_status from together.lib.cli.components.upload_progress import UploadProgressTracker, format_bytes from together.lib.cli.utils._assert_explicit_project_id import assert_explicit_project_id @@ -290,11 +291,7 @@ async def _upload_model_files( show_progress: bool = False, ) -> str: if show_progress: - with console.status( - "[progress.description]Preparing files (computing hashes)...[/progress.description]", - spinner="dots", - spinner_style="bar.pulse", - ): + with loading_status("Preparing files (computing hashes)..."): local_files = await _prepare_files(local_path) else: local_files = await _prepare_files(local_path) diff --git a/src/together/lib/cli/api/endpoints/create.py b/src/together/lib/cli/api/endpoints/create.py index d8ec04a86..72608ac0c 100644 --- a/src/together/lib/cli/api/endpoints/create.py +++ b/src/together/lib/cli/api/endpoints/create.py @@ -10,7 +10,7 @@ from together.lib.cli.utils._exit import CliDiagnosticExit from together.lib.cli.utils.config import CLIConfigParameter from together.lib.cli.utils._console import console -from together.lib.cli.components.loader import show_loading_status +from together.lib.cli.components.loader import loading_status, show_loading_status from together.lib.cli.api.endpoints._utils import print_endpoint, handle_endpoint_api_errors from .hardware import hardware as list_hardware @@ -133,11 +133,7 @@ async def create( print_endpoint(response) if wait: - with console.status( - "[progress.description]Waiting for endpoint to start...[/progress.description]", - spinner="dots", - spinner_style="bar.pulse", - ): + with loading_status("Waiting for endpoint to start..."): while (await config.client.endpoints.retrieve(response.id)).state != "STARTED": await asyncio.sleep(1) console.print("[green]√[/green] Endpoint started") diff --git a/src/together/lib/cli/api/endpoints/start.py b/src/together/lib/cli/api/endpoints/start.py index 98625b69a..d2548450a 100644 --- a/src/together/lib/cli/api/endpoints/start.py +++ b/src/together/lib/cli/api/endpoints/start.py @@ -8,7 +8,7 @@ from together._utils._json import openapi_dumps from together.lib.cli.utils.config import CLIConfigParameter from together.lib.cli.utils._console import console -from together.lib.cli.components.loader import show_loading_status +from together.lib.cli.components.loader import loading_status, show_loading_status async def start( @@ -31,11 +31,7 @@ async def start( if wait: console.print("[green]√[/green] Successfully requested endpoint to start.") - with console.status( - "[progress.description]Waiting for endpoint to start...[/progress.description]", - spinner="dots", - spinner_style="bar.pulse", - ): + with loading_status("Waiting for endpoint to start..."): while (await config.client.endpoints.retrieve(endpoint_id)).state != "STARTED": await asyncio.sleep(1) console.print("[green]√[/green] Endpoint started") diff --git a/src/together/lib/cli/api/endpoints/stop.py b/src/together/lib/cli/api/endpoints/stop.py index d160eb14b..5eb52861b 100644 --- a/src/together/lib/cli/api/endpoints/stop.py +++ b/src/together/lib/cli/api/endpoints/stop.py @@ -8,7 +8,7 @@ from together._utils._json import openapi_dumps from together.lib.cli.utils.config import CLIConfigParameter from together.lib.cli.utils._console import console -from together.lib.cli.components.loader import show_loading_status +from together.lib.cli.components.loader import loading_status, show_loading_status from together.lib.cli.api.endpoints._utils import handle_endpoint_api_errors @@ -28,11 +28,7 @@ async def stop( if wait: console.print("[green]√[/green] Successfully requested endpoint to stop.") - with console.status( - "[progress.description]Waiting for endpoint to stop...[/progress.description]", - spinner="dots", - spinner_style="bar.pulse", - ): + with loading_status("Waiting for endpoint to stop..."): while (await config.client.endpoints.retrieve(endpoint_id)).state != "STOPPED": await asyncio.sleep(1) console.print("[green]√[/green] Endpoint stopped") diff --git a/src/together/lib/cli/components/check_progress.py b/src/together/lib/cli/components/check_progress.py index c2fd6bcd3..b5b8266fa 100644 --- a/src/together/lib/cli/components/check_progress.py +++ b/src/together/lib/cli/components/check_progress.py @@ -57,7 +57,9 @@ def __init__(self, file: Path, *, enabled: bool) -> None: self._total = 1 def __enter__(self) -> CheckProgressTracker: - if not self.enabled: + from together.lib.cli.utils._debug import is_enabled + + if not self.enabled or is_enabled(): return self total = _expected_check_work_bytes(self.file) self._total = total diff --git a/src/together/lib/cli/components/download_progress.py b/src/together/lib/cli/components/download_progress.py index 93331553e..e04d9ceab 100644 --- a/src/together/lib/cli/components/download_progress.py +++ b/src/together/lib/cli/components/download_progress.py @@ -60,7 +60,9 @@ def for_single_file( ) def __enter__(self) -> DownloadProgressTracker: - if not self.enabled or not console.is_terminal: + from together.lib.cli.utils._debug import is_enabled + + if not self.enabled or not console.is_terminal or is_enabled(): return self self._progress = Progress( SpinnerColumn(style="bar.pulse"), @@ -121,16 +123,27 @@ async def file_completed(self, file_path: str, *, skipped: bool = False) -> None self.completed_files += 1 if skipped: self.skipped_files += 1 - if not self.enabled or self._progress is None: + if not self._should_report(): return - if self.show_files: + if self._progress is not None and self.show_files: assert self._files_task is not None self._progress.update( self._files_task, completed=self.completed_files, description=f"Files ({self.completed_files}/{self.total_files})", ) + if self.show_files: if skipped: - self._progress.console.print(f"[dim]↷[/dim] {file_path} skipped (already exists)") + console.print(f"[dim]↷[/dim] {file_path} skipped (already exists)") else: - self._progress.console.print(f"[success]✓[/success] {file_path} complete") + console.print(f"[success]✓[/success] {file_path} complete") + + def _should_report(self) -> bool: + """Live bar is skipped under --debug / non-TTY; still emit ✓/↷ lines in debug.""" + if not self.enabled: + return False + if self._progress is not None: + return True + from together.lib.cli.utils._debug import is_enabled + + return is_enabled() diff --git a/src/together/lib/cli/components/loader.py b/src/together/lib/cli/components/loader.py index 2ced554fa..979f60958 100644 --- a/src/together/lib/cli/components/loader.py +++ b/src/together/lib/cli/components/loader.py @@ -1,16 +1,34 @@ from __future__ import annotations -from typing import TypeVar, Awaitable +from typing import TypeVar, Iterator, Awaitable +from contextlib import contextmanager +from together.lib.cli.utils._debug import is_enabled, log_debug_note from together.lib.cli.utils._console import console T = TypeVar("T") -async def show_loading_status(message: str, request: Awaitable[T]) -> T: +@contextmanager +def loading_status(message: str) -> Iterator[None]: + """Show a spinner, or a debug line when ``--debug`` is on. + + Rich Live (stdout) and debug logs (stderr) share the terminal cursor, so a + spinner would overwrite HTTP debug lines. Skip Live UI in debug mode. + """ + if is_enabled(): + if message: + log_debug_note(message) + yield + return with console.status( f"[progress.description]{message}[/progress.description]", spinner="dots", spinner_style="bar.pulse", ): + yield + + +async def show_loading_status(message: str, request: Awaitable[T]) -> T: + with loading_status(message): return await request diff --git a/src/together/lib/cli/components/upload_progress.py b/src/together/lib/cli/components/upload_progress.py index 0b21f1a69..c902a2c1d 100644 --- a/src/together/lib/cli/components/upload_progress.py +++ b/src/together/lib/cli/components/upload_progress.py @@ -113,7 +113,9 @@ def from_upload_plan( ) def __enter__(self) -> UploadProgressTracker: - if not self.enabled or not console.is_terminal: + from together.lib.cli.utils._debug import is_enabled + + if not self.enabled or not console.is_terminal or is_enabled(): return self self._progress = Progress( SpinnerColumn(style="bar.pulse"), @@ -173,18 +175,20 @@ async def part_completed( async with self._lock: self.uploaded_bytes += bytes_count self.completed_parts += 1 - if not self.enabled or self._progress is None: + if not self._should_report(): return - assert self._bytes_task is not None - self._progress.update(self._bytes_task, completed=self.uploaded_bytes) + if self._progress is not None: + assert self._bytes_task is not None + self._progress.update(self._bytes_task, completed=self.uploaded_bytes) + if self.show_parts: + assert self._parts_task is not None + self._progress.update( + self._parts_task, + completed=self.completed_parts, + description=f"Parts ({self.completed_parts}/{self.total_parts})", + ) if self.show_parts: - assert self._parts_task is not None - self._progress.update( - self._parts_task, - completed=self.completed_parts, - description=f"Parts ({self.completed_parts}/{self.total_parts})", - ) - self._progress.console.print( + console.print( f"[success]✓[/success] {file_path} part {part_number}/{total_file_parts} " f"({format_bytes(bytes_count)})" ) @@ -192,16 +196,27 @@ async def part_completed( async def file_completed(self, file_path: str) -> None: async with self._lock: self.completed_files += 1 - if not self.enabled or self._progress is None: + if not self._should_report(): return - if self.show_files: + if self._progress is not None and self.show_files: assert self._files_task is not None self._progress.update( self._files_task, completed=self.completed_files, description=f"Files ({self.completed_files}/{self.total_files})", ) - self._progress.console.print(f"[success]✓[/success] {file_path} complete") + if self.show_files: + console.print(f"[success]✓[/success] {file_path} complete") + + def _should_report(self) -> bool: + """Live bar is skipped under --debug / non-TTY; still emit ✓ lines in debug.""" + if not self.enabled: + return False + if self._progress is not None: + return True + from together.lib.cli.utils._debug import is_enabled + + return is_enabled() async def upload_file_with_progress( diff --git a/src/together/lib/cli/utils/_debug.py b/src/together/lib/cli/utils/_debug.py new file mode 100644 index 000000000..3865f5e10 --- /dev/null +++ b/src/together/lib/cli/utils/_debug.py @@ -0,0 +1,382 @@ +from __future__ import annotations + +import os +import re +import time +import logging +import platform +import traceback +from typing import Union, Mapping +from collections.abc import Sequence +from typing_extensions import override + +import httpx +from rich.markup import escape as escape_rich_markup + +from together import __version__ +from together.lib.utils._log import set_cli_debug_console_redirect +from together.lib.cli._track_cli import _redact_secrets_in_error_text +from together.lib.cli.utils._console import error_console + +_START_EXTENSION = "together_cli_debug_start" + +_REQUEST_ID_HEADERS = ( + "x-request-id", + "x-together-request-id", + "cf-ray", + "x-amzn-requestid", + "x-amzn-trace-id", + "traceparent", + "x-cloud-trace-context", +) + +_NOISY_LOG_PATTERNS = ( + re.compile(r"^Request options:"), + re.compile(r"^Sending HTTP Request:"), + re.compile(r"^HTTP Response:"), + re.compile(r"^HTTP Request:"), + re.compile(r"Analytics event sending"), + re.compile(r"Analytics tracking disabled"), + re.compile(r"Error tracking api request"), + re.compile(r"Updating hash with chunk"), + re.compile(r"^Starting file checksum"), + re.compile(r"^hash complete", re.I), + re.compile(r"^1 retry left$"), + re.compile(r"^\d+ retries left$"), + re.compile(r"^Not retrying$"), + re.compile(r"^Retrying as header"), + re.compile(r"^Not retrying as header"), + re.compile(r"^Retrying due to status code"), + re.compile(r"^Could not read JSON from response"), + re.compile(r"^Encountered httpx\.HTTPStatusError"), + re.compile(r"^Re-raising status error$"), +) + +_enabled = False +_base_url = "" +_saved_httpx_level: int | None = None +_saved_together_level: int | None = None +_saved_together_propagate: bool | None = None +# (was_set, value) for TOGETHER_LOG before setup; None means setup did not snapshot yet. +_saved_together_log_env: tuple[bool, str] | None = None + + +def is_enabled() -> bool: + return _enabled + + +def mask_secret(value: str, *, visible: int = 4) -> str: + if not value: + return "" + if len(value) <= visible: + return "" + return f"…{value[-visible:]}" + + +def extract_request_id(headers: Mapping[str, str] | httpx.Headers) -> str | None: + lowered = {str(key).lower(): str(value) for key, value in headers.items()} + for name in _REQUEST_ID_HEADERS: + value = lowered.get(name) + if value: + return value + for key, value in lowered.items(): + if "request-id" in key or key.endswith("-trace-id"): + return value + return None + + +def is_noisy_log_message(message: str) -> bool: + text = message.strip() + return any(pattern.search(text) for pattern in _NOISY_LOG_PATTERNS) + + +_URL_WITH_QUERY_RE = re.compile(r"(https?://[^\s?#]+)(\?[^\s]*)", re.IGNORECASE) + + +def sanitize_debug_log_message(message: str) -> str: + """Redact secrets and drop URL query strings (presigned S3, SigV4, tokens).""" + stripped = _URL_WITH_QUERY_RE.sub(r"\1", message) + return _redact_secrets_in_error_text(stripped) + + +def _safe_url(url: httpx.URL, *, base_url: str = "") -> str: + if url.query: + url = url.copy_with(query=None) + rendered = str(url) + base = base_url.rstrip("/") + if base and rendered.startswith(base): + rest = rendered[len(base) :] + return rest if rest.startswith("/") else f"/{rest}" + return rendered + + +def _format_duration(seconds: float) -> str: + ms = seconds * 1000 + if ms < 10: + return f"{ms:.1f}ms" + if ms < 1000: + return f"{ms:.0f}ms" + return f"{seconds:.2f}s" + + +def _format_timeout(timeout: float | httpx.Timeout | None) -> str: + if timeout is None: + return "off" + if isinstance(timeout, (int, float)): + return f"{timeout:g}s" + read = timeout.read + if read is None: + return "off" + return f"{read:g}s" + + +def _status_style(status_code: int) -> str: + if status_code < 300: + return "success" + if status_code < 400: + return "info" + if status_code < 500: + return "warning" + return "error" + + +def render_session_lines( + *, + command: str, + is_beta_command: bool, + base_url: str, + project_id: str | None, + api_key: str | None, + timeout: float | httpx.Timeout | None, + max_retries: int, +) -> list[str]: + path = command.strip() + if is_beta_command and path: + path = f"beta {path}" + elif is_beta_command: + path = "beta" + invocation = f"tg {path}".rstrip() + + key_display = mask_secret(api_key) if api_key else "" + project_display = project_id or "" + runtime = f"python {platform.python_version()} {platform.system().lower()}" + + return [ + f"[primary]tg {escape_rich_markup(__version__)}[/primary] [dim]{escape_rich_markup(runtime)}[/dim]", + f"[bold]{escape_rich_markup(invocation)}[/bold]", + f"[dim]{escape_rich_markup(base_url)}[/dim]", + ( + f"project={escape_rich_markup(project_display)} " + f"key={escape_rich_markup(key_display)} " + f"timeout={escape_rich_markup(_format_timeout(timeout))} " + f"retries={max_retries}" + ), + ] + + +def render_request_lines(request: httpx.Request, *, base_url: str = "") -> list[str]: + method = request.method.upper() + url = _safe_url(request.url, base_url=base_url) + retry = request.headers.get("x-stainless-retry-count") + retry_bit = "" + if retry and retry != "0": + retry_bit = f" [warning]retry {escape_rich_markup(retry)}[/warning]" + + return [f"[info]→ {escape_rich_markup(method)}[/info] [bold]{escape_rich_markup(url)}[/bold]{retry_bit}"] + + +def render_response_lines(response: httpx.Response, *, elapsed: float | None = None) -> list[str]: + status = f"{response.status_code} {response.reason_phrase}".strip() + style = _status_style(response.status_code) + extras: list[str] = [] + if elapsed is not None: + extras.append(f"[dim]{_format_duration(elapsed)}[/dim]") + request_id = extract_request_id(response.headers) + if request_id: + extras.append(f"[muted]{escape_rich_markup(request_id)}[/muted]") + + suffix = (" " + " ".join(extras)) if extras else "" + return [f"[{style}]← {escape_rich_markup(status)}[/{style}]{suffix}"] + + +def _debug_print(markup: str = "") -> None: + # stderr-to-file is not a TTY, so Rich would otherwise wrap at 80 columns + # and split paths / traceback lines mid-token. + error_console.print(markup, soft_wrap=True) + + +def _print_lines(lines: Sequence[str]) -> None: + for line in lines: + _debug_print(f"[muted]debug[/muted] {line}") + + +def log_debug_session( + *, + command: str, + is_beta_command: bool, + base_url: str, + project_id: str | None, + api_key: str | None, + timeout: float | httpx.Timeout | None, + max_retries: int, +) -> None: + global _base_url + _base_url = str(base_url) + _print_lines( + render_session_lines( + command=command, + is_beta_command=is_beta_command, + base_url=str(base_url), + project_id=project_id, + api_key=api_key, + timeout=timeout, + max_retries=max_retries, + ) + ) + + +def log_debug_note(message: str) -> None: + _debug_print(f"[muted]debug[/muted] {escape_rich_markup(message)}") + + +async def _on_request(request: httpx.Request) -> None: + if not _enabled: + return + request.extensions[_START_EXTENSION] = time.perf_counter() + _print_lines(render_request_lines(request, base_url=_base_url)) + + +async def _on_response(response: httpx.Response) -> None: + if not _enabled: + return + start = response.request.extensions.get(_START_EXTENSION) + elapsed: float | None + if isinstance(start, (int, float)): + elapsed = time.perf_counter() - float(start) + else: + elapsed = None + _print_lines(render_response_lines(response, elapsed=elapsed)) + _debug_print() + + +class CliDebugLogFilter(logging.Filter): + @override + def filter(self, record: logging.LogRecord) -> bool: + try: + return not is_noisy_log_message(record.getMessage()) + except Exception: + return True + + +def _traceback_text(record: logging.LogRecord) -> str | None: + exc_info = record.exc_info + if not exc_info or exc_info[0] is None: + return None + return "".join(traceback.format_exception(*exc_info)).rstrip() + + +class CliDebugLogHandler(logging.Handler): + @override + def emit(self, record: logging.LogRecord) -> None: + if not _enabled: + return + try: + message = sanitize_debug_log_message(record.getMessage()) + level = record.levelname.lower() + style = { + "debug": "muted", + "info": "info", + "warning": "warning", + "error": "error", + "critical": "error", + }.get(level, "muted") + name = record.name.removeprefix("together.").removeprefix("together") + _debug_print( + f"[muted]log[/muted] [{style}]{escape_rich_markup(level)}[/{style}] " + f"[dim]{escape_rich_markup(name)}[/dim] {escape_rich_markup(message)}" + ) + tb = _traceback_text(record) + if tb: + _debug_print(f"[muted]log[/muted] [dim]{escape_rich_markup(sanitize_debug_log_message(tb))}[/dim]") + except Exception: + self.handleError(record) + + +def install_http_debug_hooks(http_client: httpx.AsyncClient) -> None: + hooks = http_client.event_hooks + request_hooks = hooks.setdefault("request", []) + response_hooks = hooks.setdefault("response", []) + if _on_request not in request_hooks: + request_hooks.append(_on_request) + if _on_response not in response_hooks: + response_hooks.append(_on_response) + + +def setup_cli_debug_logging() -> None: + global _enabled, _saved_httpx_level, _saved_together_level, _saved_together_propagate, _saved_together_log_env + if _saved_together_log_env is None: + env_value = os.environ.get("TOGETHER_LOG") + _saved_together_log_env = ("TOGETHER_LOG" in os.environ, env_value or "") + os.environ.setdefault("TOGETHER_LOG", "debug") + _enabled = True + set_cli_debug_console_redirect(True) + + httpx_logger = logging.getLogger("httpx") + together_logger = logging.getLogger("together") + _saved_httpx_level = httpx_logger.level + _saved_together_level = together_logger.level + _saved_together_propagate = together_logger.propagate + + httpx_logger.setLevel(logging.WARNING) + together_logger.setLevel(logging.DEBUG) + together_logger.propagate = False + + if not any(isinstance(handler, CliDebugLogHandler) for handler in together_logger.handlers): + handler = CliDebugLogHandler() + handler.setLevel(logging.DEBUG) + handler.addFilter(CliDebugLogFilter()) + together_logger.addHandler(handler) + + +def teardown_cli_debug() -> None: + global \ + _enabled, \ + _base_url, \ + _saved_httpx_level, \ + _saved_together_level, \ + _saved_together_propagate, \ + _saved_together_log_env + _enabled = False + _base_url = "" + set_cli_debug_console_redirect(False) + + together_logger = logging.getLogger("together") + together_logger.handlers = [ + handler for handler in together_logger.handlers if not isinstance(handler, CliDebugLogHandler) + ] + if _saved_together_propagate is not None: + together_logger.propagate = _saved_together_propagate + _saved_together_propagate = None + if _saved_together_level is not None: + together_logger.setLevel(_saved_together_level) + _saved_together_level = None + + if _saved_httpx_level is not None: + logging.getLogger("httpx").setLevel(_saved_httpx_level) + _saved_httpx_level = None + + if _saved_together_log_env is not None: + was_set, value = _saved_together_log_env + if was_set: + os.environ["TOGETHER_LOG"] = value + else: + os.environ.pop("TOGETHER_LOG", None) + _saved_together_log_env = None + + +def format_timeout_for_display(timeout: Union[float, httpx.Timeout, None]) -> str: + return _format_timeout(timeout) + + +def format_duration_for_display(seconds: float) -> str: + return _format_duration(seconds) diff --git a/src/together/lib/utils/_log.py b/src/together/lib/utils/_log.py index c7944d402..3bf6289d5 100644 --- a/src/together/lib/utils/_log.py +++ b/src/together/lib/utils/_log.py @@ -12,12 +12,21 @@ WARNING_MESSAGES_ONCE: Set[str] = set() +# When the CLI --debug handler is attached, skip the raw print() path so messages +# are not duplicated (and so they go through the formatted/redacted handler). +_cli_debug_console_redirect = False + + +def set_cli_debug_console_redirect(enabled: bool) -> None: + global _cli_debug_console_redirect + _cli_debug_console_redirect = enabled + def _console_log_level() -> str | None: - if TOGETHER_LOG in ["debug", "info"]: - return TOGETHER_LOG - else: - return None + env = os.environ.get("TOGETHER_LOG", TOGETHER_LOG) + if env in ["debug", "info"]: + return env + return None def logfmt(props: Dict[str, Any]) -> str: @@ -40,27 +49,29 @@ def fmt(key: str, val: Any) -> str: def log_debug(message: str | Any, **params: Any) -> None: msg = logfmt(dict(message=message, **params)) - if _console_log_level() == "debug": + if _console_log_level() == "debug" and not _cli_debug_console_redirect: print(msg, file=sys.stderr) # noqa logger.debug(msg) def log_info(message: str | Any, **params: Any) -> None: msg = logfmt(dict(message=message, **params)) - if _console_log_level() in ["debug", "info"]: + if _console_log_level() in ["debug", "info"] and not _cli_debug_console_redirect: print(msg, file=sys.stderr) # noqa logger.info(msg) def log_warn(message: str | Any, **params: Any) -> None: msg = logfmt(dict(message=message, **params)) - print(msg, file=sys.stderr) # noqa + if not _cli_debug_console_redirect: + print(msg, file=sys.stderr) # noqa logger.warning(msg) def log_warn_once(message: str | Any, **params: Any) -> None: msg = logfmt(dict(message=message, **params)) if msg not in WARNING_MESSAGES_ONCE: - print(msg, file=sys.stderr) # noqa + if not _cli_debug_console_redirect: + print(msg, file=sys.stderr) # noqa logger.warning(msg) WARNING_MESSAGES_ONCE.add(msg) diff --git a/tests/cli/test_debug.py b/tests/cli/test_debug.py new file mode 100644 index 000000000..69b8326be --- /dev/null +++ b/tests/cli/test_debug.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import os +import json +import logging + +import httpx +import pytest +from respx import MockRouter + +from together import APIError +from tests.cli.utils import API_KEY, CliRunner +from together._version import __version__ + +base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") + + +def _whoami_body() -> dict[str, str]: + return { + "api_key_id": "key-1", + "organization_id": "org-1", + "organization_name": "Acme Org", + "project_id": "proj", + "project_name": "My Project", + "project_slug": "my-project", + "user_id": "user-1", + } + + +class TestCliDebug: + def test_help_lists_debug_flag(self, cli_runner: CliRunner) -> None: + result = cli_runner.invoke(["--help"]) + assert result.exit_code == 0 + assert "--debug" in result.output + + @pytest.mark.respx(base_url=base_url) + def test_debug_whoami_is_structured_and_redacted(self, respx_mock: MockRouter, cli_runner: CliRunner) -> None: + respx_mock.get("/whoami").mock( + return_value=httpx.Response( + 200, + json=_whoami_body(), + headers={"x-request-id": "req_test_123", "server": "cloudflare"}, + ) + ) + + result = cli_runner.invoke(["whoami", "--debug"]) + + assert result.exit_code == 0, result.output + assert "My Project" in result.out_out + err = result.err_out + assert "debug" in err + assert "debug debug" not in err + assert __version__ in err + assert "tg whoami" in err + assert "GET" in err + assert "200" in err + assert "req_test_123" in err + assert "api_key_id" not in err + assert "organization_id" not in err + assert API_KEY not in err + assert "Request options:" not in err + assert "Sending HTTP Request:" not in err + assert "HTTP Response:" not in err + assert "Headers(" not in err + assert "Analytics event sending" not in err + assert "server: cloudflare" not in err.lower() + + @pytest.mark.respx(base_url=base_url) + def test_debug_keeps_json_stdout_clean(self, respx_mock: MockRouter, cli_runner: CliRunner) -> None: + respx_mock.get("/whoami").mock(return_value=httpx.Response(200, json=_whoami_body())) + + result = cli_runner.invoke(["whoami", "--json", "--debug"]) + + assert result.exit_code == 0, result.output + payload = json.loads(result.out_out) + assert payload["project_id"] == "proj" + assert "GET" in result.err_out + assert "debug" in result.err_out + + @pytest.mark.respx(base_url=base_url) + def test_debug_error_keeps_status_without_response_body( + self, respx_mock: MockRouter, cli_runner: CliRunner + ) -> None: + respx_mock.get("/whoami").mock( + return_value=httpx.Response( + 401, + json={"error": {"message": "Invalid API key", "type": "invalid_request_error"}}, + headers={"x-request-id": "req_err_1"}, + ) + ) + + with pytest.raises(APIError): + cli_runner.invoke(["whoami", "--debug"]) + captured = cli_runner.capsys.readouterr() + err = captured.err + assert "401" in err + assert "req_err_1" in err + assert "invalid_request_error" not in err + assert API_KEY not in err + assert "Invalid API key" in captured.out + + @pytest.mark.respx(base_url=base_url) + def test_debug_off_does_not_print_http_trace(self, respx_mock: MockRouter, cli_runner: CliRunner) -> None: + respx_mock.get("/whoami").mock(return_value=httpx.Response(200, json=_whoami_body())) + + result = cli_runner.invoke(["whoami"]) + + assert result.exit_code == 0, result.output + assert "→" not in result.err_out + assert "tg whoami" not in result.err_out + + def test_debug_missing_api_key_shows_missing_and_request( + self, cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("TOGETHER_API_KEY", raising=False) + cli_runner.env.pop("TOGETHER_API_KEY", None) + tracked: list[object] = [] + + def _spy(event: object, _args: object) -> None: + tracked.append(event) + + monkeypatch.setattr("together.lib.cli.track_cli", _spy) + + result = cli_runner.invoke(["whoami", "--debug"]) + + assert result.exit_code == 1 + err = result.err_out + assert "key=" in err + assert "…0000" not in err + assert "GET" in err + assert "api key missing" in result.out_out.lower() + from together.lib.cli._track_cli import CliTrackingEvents + + assert CliTrackingEvents.ApiRequest not in tracked + + @pytest.mark.respx(base_url=base_url) + def test_debug_does_not_leave_logger_hooked(self, respx_mock: MockRouter, cli_runner: CliRunner) -> None: + respx_mock.get("/whoami").mock(return_value=httpx.Response(200, json=_whoami_body())) + + first = cli_runner.invoke(["whoami", "--debug"]) + assert first.exit_code == 0, first.output + + together_logger = logging.getLogger("together") + from together.lib.cli.utils._debug import CliDebugLogHandler, is_enabled + + assert is_enabled() is False + assert not any(isinstance(handler, CliDebugLogHandler) for handler in together_logger.handlers) + + second = cli_runner.invoke(["whoami"]) + assert second.exit_code == 0, second.output + assert "→" not in second.err_out diff --git a/tests/unit/test_cli_debug.py b/tests/unit/test_cli_debug.py new file mode 100644 index 000000000..6e87cecc0 --- /dev/null +++ b/tests/unit/test_cli_debug.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import os +import logging +from contextlib import contextmanager + +import httpx +import pytest + +from together.lib.cli.utils._debug import ( + CliDebugLogFilter, + mask_secret, + log_debug_note, + extract_request_id, + teardown_cli_debug, + is_noisy_log_message, + render_request_lines, + render_session_lines, + render_response_lines, + setup_cli_debug_logging, + sanitize_debug_log_message, + format_duration_for_display, +) + + +def test_mask_secret_keeps_only_tail() -> None: + assert mask_secret("abcdefghijklmnop") == "…mnop" + assert mask_secret("abcd") == "" + assert mask_secret("") == "" + + +def test_extract_request_id_prefers_x_request_id() -> None: + headers = httpx.Headers({"cf-ray": "ray-1", "x-request-id": "req_abc"}) + assert extract_request_id(headers) == "req_abc" + + +def test_noisy_sdk_and_analytics_messages_are_dropped() -> None: + assert is_noisy_log_message("Request options: {'headers': {'Authorization': 'Bearer x'}}") + assert is_noisy_log_message('HTTP Response: GET https://api.together.ai/v1/whoami "200 OK" Headers(...)') + assert is_noisy_log_message("Sending HTTP Request: GET https://api.together.ai/v1/whoami") + assert is_noisy_log_message("Analytics event sending") + assert is_noisy_log_message("Updating hash with chunk of size 8192") + assert is_noisy_log_message("Encountered httpx.HTTPStatusError") + assert not is_noisy_log_message("Retrying request to /whoami in 1.000000 seconds") + assert not is_noisy_log_message("Raising timeout error") + + +def test_log_filter_uses_interpolated_message() -> None: + log_filter = CliDebugLogFilter() + noisy = logging.LogRecord( + "together._base_client", + logging.DEBUG, + __file__, + 1, + "Request options: %s", + ({"headers": "secret"},), + None, + ) + useful = logging.LogRecord( + "together._base_client", + logging.INFO, + __file__, + 1, + "Retrying request to %s in %f seconds", + ("/whoami", 1.0), + None, + ) + assert log_filter.filter(noisy) is False + assert log_filter.filter(useful) is True + + +def test_request_render_is_method_and_path_only() -> None: + request = httpx.Request( + "POST", + "https://api.together.ai/v1/fine-tunes?api_key=secretvalue&foo=bar", + headers={ + "Authorization": "Bearer supersecretapikeyvalue", + "Content-Type": "application/json", + "X-Stainless-Lang": "python", + "X-Stainless-Retry-Count": "2", + "Accept-Encoding": "gzip", + }, + content=b'{"model":"demo"}', + ) + blob = "\n".join(render_request_lines(request, base_url="https://api.together.ai/v1")) + assert "→ POST" in blob + assert "/fine-tunes" in blob + assert "foo=bar" not in blob + assert "secretvalue" not in blob + assert "supersecretapikeyvalue" not in blob + assert "retry 2" in blob + assert "authorization" not in blob.lower() + assert "content-type" not in blob.lower() + assert "demo" not in blob + + +def test_response_render_is_status_line_only() -> None: + request = httpx.Request("GET", "https://api.together.ai/v1/whoami") + response = httpx.Response( + 200, + json={"project_id": "proj", "organization_id": "org"}, + headers={ + "x-request-id": "req_test_123", + "content-type": "application/json", + "date": "Wed, 01 Jan 2024 00:00:00 GMT", + "server": "cloudflare", + "cf-ray": "should-not-duplicate-if-request-id-present", + }, + request=request, + ) + blob = "\n".join(render_response_lines(response, elapsed=0.048)) + assert "← 200" in blob + assert "req_test_123" in blob + assert format_duration_for_display(0.048) in blob + assert "project_id" not in blob + assert "organization_id" not in blob + assert "cloudflare" not in blob.lower() + assert "content-type" not in blob.lower() + assert "cf-ray" not in blob.lower() + + +def test_session_banner_missing_key() -> None: + blob = "\n".join( + render_session_lines( + command="whoami", + is_beta_command=False, + base_url="https://api.together.ai/v1/", + project_id=None, + api_key=None, + timeout=None, + max_retries=0, + ) + ) + assert "key=" in blob + assert "project=" in blob + + +def test_session_banner_masks_api_key() -> None: + blob = "\n".join( + render_session_lines( + command="whoami", + is_beta_command=False, + base_url="https://api.together.ai/v1/", + project_id="proj", + api_key="abcdefghijklmnop", + timeout=60, + max_retries=0, + ) + ) + assert "tg whoami" in blob + assert "abcdefghijklmnop" not in blob + assert "…mnop" in blob + assert "proj" in blob + assert "retries=0" in blob + assert "[muted]debug[/muted]" not in blob + + +async def test_show_loading_status_skips_spinner_when_debug(monkeypatch: pytest.MonkeyPatch) -> None: + from together.lib.cli.utils import _debug + from together.lib.cli.utils._console import console + from together.lib.cli.components.loader import show_loading_status + + monkeypatch.setattr(_debug, "_enabled", True) + + def fail_status(*_args: object, **_kwargs: object) -> None: + raise AssertionError("spinner should be skipped in debug mode") + + monkeypatch.setattr(console, "status", fail_status) + + async def done() -> str: + return "ok" + + assert await show_loading_status("Loading widgets...", done()) == "ok" + + +async def test_show_loading_status_uses_spinner_when_not_debug(monkeypatch: pytest.MonkeyPatch) -> None: + from together.lib.cli.utils import _debug + from together.lib.cli.utils._console import console + from together.lib.cli.components.loader import show_loading_status + + monkeypatch.setattr(_debug, "_enabled", False) + used = {"status": False} + + @contextmanager + def fake_status(*_args: object, **_kwargs: object): + used["status"] = True + yield + + monkeypatch.setattr(console, "status", fake_status) + + async def done() -> int: + return 1 + + assert await show_loading_status("Loading...", done()) == 1 + assert used["status"] is True + + +def test_sanitize_debug_log_strips_presigned_query() -> None: + url = ( + "https://s3.amazonaws.com/bucket/key?X-Amz-Algorithm=AWS4-HMAC-SHA256" + "&X-Amz-Credential=AKIAEXAMPLE%2F20260101%2Fus-east-1%2Fs3%2Faws4_request" + "&X-Amz-Signature=deadbeefcafebabe" + ) + out = sanitize_debug_log_message(f"Upload redirected to {url}") + assert out.startswith("Upload redirected to https://s3.amazonaws.com/bucket/key") + assert "X-Amz-" not in out + assert "AKIAEXAMPLE" not in out + assert "deadbeef" not in out + assert "?" not in out + + +def test_teardown_restores_together_log_env_and_logger_level( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.delenv("TOGETHER_LOG", raising=False) + together_logger = logging.getLogger("together") + previous_level = together_logger.level + together_logger.setLevel(logging.WARNING) + try: + setup_cli_debug_logging() + assert os.environ.get("TOGETHER_LOG") == "debug" + assert together_logger.level == logging.DEBUG + teardown_cli_debug() + assert "TOGETHER_LOG" not in os.environ + assert together_logger.level == logging.WARNING + + from together.lib.utils._log import log_debug + + log_debug("Analytics event sending", body="should-not-print") + captured = capsys.readouterr() + assert "Analytics event sending" not in captured.err + assert "should-not-print" not in captured.err + finally: + teardown_cli_debug() + together_logger.setLevel(previous_level) + + +def test_log_warn_not_duplicated_under_cli_debug( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.delenv("TOGETHER_LOG", raising=False) + setup_cli_debug_logging() + try: + from together.lib.utils._log import log_warn, log_warn_once + + log_warn("validation loops disabled") + log_warn_once("cli-debug-warn-once-unique") + err = capsys.readouterr().err + assert err.count("validation loops disabled") == 1 + assert err.count("cli-debug-warn-once-unique") == 1 + assert "message=" in err + finally: + teardown_cli_debug() + + +def test_cli_debug_handler_prints_traceback(capsys: pytest.CaptureFixture[str]) -> None: + setup_cli_debug_logging() + try: + logger = logging.getLogger("together._base_client") + try: + raise TimeoutError("connect timed out") + except TimeoutError: + logger.debug("Encountered httpx.TimeoutException", exc_info=True) + err = capsys.readouterr().err + assert "Encountered httpx.TimeoutException" in err + assert "TimeoutError" in err + assert "connect timed out" in err + assert "Traceback" in err + finally: + teardown_cli_debug() + + +def test_debug_output_does_not_hard_wrap_at_80_columns( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + from together.lib.cli.utils._console import error_console + + monkeypatch.delenv("COLUMNS", raising=False) + previous_width = error_console.width + error_console.width = 80 + long_path = "/home/runner/work/together-py/.venv/lib/python3.10/site-packages/httpx/_transports/default.py" + assert len(long_path) > 80 + setup_cli_debug_logging() + try: + log_debug_note(f'File "{long_path}", line 89') + logger = logging.getLogger("together._base_client") + try: + raise TimeoutError(long_path) + except TimeoutError: + logger.debug("Encountered httpx.TimeoutException", exc_info=True) + err = capsys.readouterr().err + assert long_path in err + assert "_trans\n" not in err + assert "_trans\r" not in err + finally: + error_console.width = previous_width + teardown_cli_debug() diff --git a/tests/unit/test_upload_progress.py b/tests/unit/test_upload_progress.py index 5dc5d618e..1923fc9a1 100644 --- a/tests/unit/test_upload_progress.py +++ b/tests/unit/test_upload_progress.py @@ -90,6 +90,66 @@ class _NonTerminalConsole: assert tracker.as_callback() is not None +async def test_upload_progress_prints_completion_under_debug(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from together.lib.cli.utils import _debug + from together.lib.cli.components import upload_progress as upload_progress_mod + from together.lib.cli.components.upload_progress import UploadProgressTracker + + monkeypatch.setattr(_debug, "_enabled", True) + printed: list[str] = [] + + class _TerminalConsole: + is_terminal = True + + def print(self, message: object, *_args: object, **_kwargs: object) -> None: + printed.append(str(message)) + + monkeypatch.setattr(upload_progress_mod, "console", _TerminalConsole()) + file = tmp_path / "weights.bin" + file.write_bytes(b"x" * 8) + tracker = UploadProgressTracker( + total_bytes=8, + total_parts=1, + total_files=1, + enabled=True, + ) + with tracker: + assert tracker._progress is None + await tracker.part_completed( + file_path="weights.bin", + part_number=1, + total_file_parts=1, + bytes_count=8, + ) + await tracker.file_completed("weights.bin") + assert any("weights.bin part 1/1" in line for line in printed) + assert any("weights.bin complete" in line for line in printed) + + +async def test_download_progress_prints_completion_under_debug(monkeypatch: pytest.MonkeyPatch) -> None: + from together.lib.cli.utils import _debug + from together.lib.cli.components import download_progress as download_progress_mod + from together.lib.cli.components.download_progress import DownloadProgressTracker + + monkeypatch.setattr(_debug, "_enabled", True) + printed: list[str] = [] + + class _TerminalConsole: + is_terminal = True + + def print(self, message: object, *_args: object, **_kwargs: object) -> None: + printed.append(str(message)) + + monkeypatch.setattr(download_progress_mod, "console", _TerminalConsole()) + tracker = DownloadProgressTracker(total_bytes=10, total_files=2, enabled=True) + with tracker: + assert tracker._progress is None + await tracker.file_completed("a.bin") + await tracker.file_completed("b.bin", skipped=True) + assert any("a.bin complete" in line for line in printed) + assert any("b.bin skipped" in line for line in printed) + + def test_download_progress_tracker_skips_render_when_not_terminal(monkeypatch: pytest.MonkeyPatch) -> None: from together.lib.cli.components import download_progress as download_progress_mod from together.lib.cli.components.download_progress import DownloadProgressTracker