diff --git a/src/together/lib/cli/__init__.py b/src/together/lib/cli/__init__.py index 249efdfd..fc768fc1 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, sanitize_cli_error_message, ) +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 @@ -141,6 +146,7 @@ def _create_client( max_retries: Optional[int], project_id: Optional[str], require_api_key: bool = True, + debug: bool = False, ) -> AsyncTogether: try: client = AsyncTogether( @@ -182,6 +188,8 @@ async def track_request(request: httpx.Request) -> None: log_debug("Error tracking api request", error=e) client._client.event_hooks["request"].append(track_request) + if debug: + install_http_debug_hooks(client._client) # Out-of-band-auth commands (e.g. `beta clusters ssh`) make no Together API # calls, so a missing key is not fatal for them. The block hook installed @@ -208,7 +216,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 +246,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 +288,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 = _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=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/utils/_debug.py b/src/together/lib/cli/utils/_debug.py new file mode 100644 index 00000000..a5dea8c8 --- /dev/null +++ b/src/together/lib/cli/utils/_debug.py @@ -0,0 +1,332 @@ +from __future__ import annotations + +import os +import re +import time +import logging +import platform +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_propagate: bool | 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) + + +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"[muted]debug[/muted] [primary]tg {escape_rich_markup(__version__)}[/primary] [dim]{escape_rich_markup(runtime)}[/dim]", + f"[muted]debug[/muted] [bold]{escape_rich_markup(invocation)}[/bold]", + f"[muted]debug[/muted] [dim]{escape_rich_markup(base_url)}[/dim]", + ( + f"[muted]debug[/muted] 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 _print_lines(lines: Sequence[str]) -> None: + for line in lines: + error_console.print(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: + error_console.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)) + error_console.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 + + +class CliDebugLogHandler(logging.Handler): + @override + def emit(self, record: logging.LogRecord) -> None: + if not _enabled: + return + try: + message = _redact_secrets_in_error_text(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") + error_console.print( + f"[muted]log[/muted] [{style}]{escape_rich_markup(level)}[/{style}] " + f"[dim]{escape_rich_markup(name)}[/dim] {escape_rich_markup(message)}" + ) + except Exception: + self.handleError(record) + + +def install_http_debug_hooks(http_client: httpx.AsyncClient | httpx.Client) -> 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_propagate + 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_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_propagate + _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_httpx_level is not None: + logging.getLogger("httpx").setLevel(_saved_httpx_level) + _saved_httpx_level = 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 c7944d40..70d18431 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,14 +49,14 @@ 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) diff --git a/tests/cli/test_debug.py b/tests/cli/test_debug.py new file mode 100644 index 00000000..52fcb32d --- /dev/null +++ b/tests/cli/test_debug.py @@ -0,0 +1,123 @@ +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 __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 + 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 + + @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 00000000..447a0901 --- /dev/null +++ b/tests/unit/test_cli_debug.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import logging + +import httpx + +from together.lib.cli.utils._debug import ( + CliDebugLogFilter, + mask_secret, + extract_request_id, + is_noisy_log_message, + render_request_lines, + render_session_lines, + render_response_lines, + 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_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