Merge pull request #39435 from BerriAI/litellm_debug_claude_session_report

feat(cli): add `lite debug claude` session report and /debug-lite slash command
This commit is contained in:
Mateo Wang 2026-09-04 20:30:47 -07:00 committed by GitHub
commit 8d4a63d2a1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 605 additions and 0 deletions

View file

@ -0,0 +1,372 @@
"""`lite debug claude`: one-shot debug report for a Claude Code session routed through the proxy.
Claude Code puts its session id in `metadata.user_id`, which the proxy lifts into
`LiteLLM_SpendLogs.session_id`. This command pulls every turn of that session, plus
the request / response bodies for failures and the most recent turns, and renders a
single markdown report that can be pasted into a bug report or handed to another agent.
"""
import json
import os
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from types import MappingProxyType
from typing import Final
import click
import requests
from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError, field_validator
from ...http_client import HTTPClient
from ._cli_context import cli_context_values
CLAUDE_DIR: Final = Path.home() / ".claude"
REPORT_DIR: Final = Path.home() / ".litellm" / "debug"
SESSION_ID_ENV: Final = "CLAUDE_CODE_SESSION_ID"
SLASH_COMMAND_NAME: Final = "debug-lite"
SLASH_COMMAND_BODY: Final = """---
description: Pull the LiteLLM debug report (spend, request, response, error) for this Claude Code session
allowed-tools: Bash(lite debug claude:*)
---
Below is the LiteLLM debug report for this Claude Code session. Summarize the failing
request(s) in a few sentences (model, error, request id) and tell me the path the full
report was saved to so I can hand it off. If nothing failed, say so.
!`lite debug claude $ARGUMENTS`
"""
@dataclass(frozen=True, slots=True)
class DebugFailure:
message: str
class ErrorInformation(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
error_code: str | None = None
error_class: str | None = None
error_message: str | None = None
llm_provider: str | None = None
class SpendLogMetadata(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
status: str | None = None
error_information: ErrorInformation | None = None
class SpendLogRow(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore", populate_by_name=True)
request_id: str
start_time: str | None = Field(default=None, alias="startTime")
end_time: str | None = Field(default=None, alias="endTime")
model: str | None = None
model_group: str | None = None
custom_llm_provider: str | None = None
api_base: str | None = None
call_type: str | None = None
status: str | None = None
spend: float = 0.0
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
metadata: SpendLogMetadata = SpendLogMetadata()
@field_validator("metadata", mode="before")
@classmethod
def _parse_metadata(cls, value: object) -> object:
if value is None:
return SpendLogMetadata()
if isinstance(value, str):
return json.loads(value) if value else SpendLogMetadata()
return value
@property
def failed(self) -> bool:
return (self.status or self.metadata.status) == "failure"
@property
def error(self) -> ErrorInformation | None:
return self.metadata.error_information
class SessionLogsPage(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
data: tuple[SpendLogRow, ...]
total: int
total_pages: int
class RequestResponsePayload(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
proxy_server_request: JsonValue = None
response: JsonValue = None
messages: JsonValue = None
_SESSION_PAGE: Final = TypeAdapter(SessionLogsPage)
_PAYLOAD: Final[TypeAdapter[RequestResponsePayload | None]] = TypeAdapter(RequestResponsePayload | None)
_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
_SESSION_PAGE_SIZE: Final = 100
_TRANSPORT_BODY_CHARS: Final = 500
_SESSION_TRANSCRIPT_STEM: Final = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
def detect_claude_session_id(env: Mapping[str, str], claude_dir: Path) -> str | None:
explicit: Final = env.get(SESSION_ID_ENV)
if explicit:
return explicit
transcripts: Final = tuple(
path for path in claude_dir.glob("projects/*/*.jsonl") if _SESSION_TRANSCRIPT_STEM.fullmatch(path.stem)
)
if not transcripts:
return None
newest: Final = max(transcripts, key=lambda p: p.stat().st_mtime)
return newest.stem
def _transport_failure(uri: str, error: requests.exceptions.RequestException) -> DebugFailure:
body: Final = error.response.text[:_TRANSPORT_BODY_CHARS] if error.response is not None else ""
detail: Final = f"\n{body}" if body else ""
return DebugFailure(f"GET {uri} failed: {error}{detail}")
class SpendLogsFetcher:
def __init__(self, http: HTTPClient) -> None:
self._http = http
def session_rows(self, session_id: str) -> tuple[SpendLogRow, ...] | DebugFailure:
first: Final = self._page(session_id, 1)
if isinstance(first, DebugFailure):
return first
rest: Final = tuple(self._page(session_id, page) for page in range(2, first.total_pages + 1))
failed_page: Final = next((page for page in rest if isinstance(page, DebugFailure)), None)
if failed_page is not None:
return failed_page
rows: Final = first.data + tuple(row for page in rest if isinstance(page, SessionLogsPage) for row in page.data)
return tuple(sorted(rows, key=lambda r: r.start_time or ""))
def _get(self, uri: str, params: Mapping[str, str | int] | None = None) -> JsonValue | DebugFailure:
try:
return _JSON.validate_python(self._http.request("GET", uri, params=params)) # pyright: ignore[reportUnknownMemberType] # HTTPClient.request is untyped
except requests.exceptions.RequestException as e:
return _transport_failure(uri, e)
def _page(self, session_id: str, page: int) -> SessionLogsPage | DebugFailure:
uri: Final = "/spend/logs/session/ui"
raw: Final = self._get(
uri, MappingProxyType({"session_id": session_id, "page": page, "page_size": _SESSION_PAGE_SIZE})
)
if isinstance(raw, DebugFailure):
return raw
try:
return _SESSION_PAGE.validate_python(raw)
except ValidationError as e:
return DebugFailure(f"Unexpected {uri} response: {e}")
def payload(self, request_id: str) -> RequestResponsePayload | None | DebugFailure:
uri: Final = f"/spend/logs/ui/{request_id}"
raw: Final = self._get(uri)
if isinstance(raw, DebugFailure):
return raw
try:
return _PAYLOAD.validate_python(raw)
except ValidationError as e:
return DebugFailure(f"Unexpected {uri} response: {e}")
def _fmt_json(value: JsonValue, max_chars: int) -> str:
text: Final = value if isinstance(value, str) else json.dumps(value, indent=2, default=str)
if len(text) <= max_chars:
return text
return f"{text[:max_chars]}\n... (truncated, {len(text) - max_chars} more chars)"
def _fenced(text: str, info: str = "") -> tuple[str, str, str]:
longest_run: Final = max((len(run) for run in re.findall(r"`+", text)), default=0)
fence: Final = "`" * max(3, longest_run + 1)
return (f"{fence}{info}", text, fence)
def _row_section(row: SpendLogRow, index: int, payload: RequestResponsePayload | None, max_chars: int) -> str:
err: Final = row.error
error_lines: Final = (
(
f"- error: `{err.error_code or '?'}` {err.error_class or ''}".rstrip(),
"",
*_fenced(err.error_message or ""),
)
if err is not None and row.failed
else ()
)
body_lines: Final = (
(
"",
"<details><summary>request body</summary>",
"",
*_fenced(_fmt_json(payload.proxy_server_request, max_chars), "json"),
"</details>",
"",
"<details><summary>response</summary>",
"",
*_fenced(_fmt_json(payload.response, max_chars), "json"),
"</details>",
)
if payload is not None
else ()
)
header: Final = f"### {index}. {'FAILED' if row.failed else 'ok'} {row.model or row.model_group or '?'}"
facts: Final = (
f"- request_id: `{row.request_id}`",
f"- time: {row.start_time} -> {row.end_time}",
f"- provider: {row.custom_llm_provider or '?'} ({row.api_base or 'n/a'}), call_type: {row.call_type or '?'}",
f"- spend: ${row.spend:.6f}, tokens: {row.prompt_tokens} in / {row.completion_tokens} out",
)
return "\n".join((header, *facts, *error_lines, *body_lines))
def render_report(
*,
session_id: str,
base_url: str,
rows: Sequence[SpendLogRow],
payloads: Mapping[str, RequestResponsePayload | None],
max_chars: int,
) -> str:
failures: Final = tuple(r for r in rows if r.failed)
summary: Final = (
f"# LiteLLM debug report: Claude Code session `{session_id}`",
"",
f"- proxy: {base_url}",
f"- generated: {datetime.now(timezone.utc).isoformat(timespec='seconds')}",
f"- turns: {len(rows)}, failed: {len(failures)}",
f"- total spend: ${sum(r.spend for r in rows):.6f}",
f"- models: {', '.join(sorted(frozenset(r.model or r.model_group or '?' for r in rows))) or 'n/a'}",
"",
"Bodies are included for failed turns and the most recent turns. "
"Bodies are empty unless the proxy runs with `general_settings.store_prompts_in_spend_logs: true`.",
"",
"## Turns",
"",
)
sections: Final = tuple(
_row_section(row, i, payloads.get(row.request_id), max_chars) for i, row in enumerate(rows, start=1)
)
return "\n".join(summary) + "\n\n".join(sections) + "\n"
def build_report(
*,
fetcher: SpendLogsFetcher,
session_id: str,
base_url: str,
recent_bodies: int,
max_chars: int,
) -> str | DebugFailure:
rows: Final = fetcher.session_rows(session_id)
if isinstance(rows, DebugFailure):
return rows
if not rows:
return DebugFailure(
f"No spend logs found for session {session_id!r} on {base_url}. "
"Is Claude Code routed through this proxy (`lite up`), and does your key have log access?"
)
wanted: Final = frozenset(r.request_id for r in rows if r.failed) | frozenset(
r.request_id for r in rows[-recent_bodies:] if recent_bodies > 0
)
fetched: Final = MappingProxyType({rid: fetcher.payload(rid) for rid in sorted(wanted)})
failed_payload: Final = next((p for p in fetched.values() if isinstance(p, DebugFailure)), None)
if failed_payload is not None:
return failed_payload
payloads: Final = MappingProxyType({rid: p for rid, p in fetched.items() if not isinstance(p, DebugFailure)})
return render_report(session_id=session_id, base_url=base_url, rows=rows, payloads=payloads, max_chars=max_chars)
def write_report(report: str, session_id: str, report_dir: Path) -> Path:
report_dir.mkdir(parents=True, exist_ok=True)
path: Final = report_dir / f"claude-{session_id}.md"
path.write_text(report, encoding="utf-8")
path.chmod(0o600)
return path
def install_slash_command(claude_dir: Path) -> Path:
commands_dir: Final = claude_dir / "commands"
commands_dir.mkdir(parents=True, exist_ok=True)
path: Final = commands_dir / f"{SLASH_COMMAND_NAME}.md"
path.write_text(SLASH_COMMAND_BODY, encoding="utf-8")
return path
@click.group()
def debug() -> None:
"""Pull debug reports (spend, request, response, error) for coding-agent sessions"""
@debug.command("claude")
@click.option(
"--session-id",
default=None,
help=f"Claude Code session id. Defaults to ${SESSION_ID_ENV}, else the most recently used transcript in ~/.claude",
)
@click.option(
"--recent-bodies",
default=3,
show_default=True,
type=click.IntRange(min=0),
help="Also include request/response bodies for the N most recent turns (failed turns always get bodies)",
)
@click.option(
"--max-body-chars",
default=20_000,
show_default=True,
type=click.IntRange(min=100),
help="Truncate each request/response body to this many characters",
)
@click.option("--no-save", is_flag=True, help="Print only, do not write the report under ~/.litellm/debug")
@click.pass_context
def debug_claude(
ctx: click.Context, session_id: str | None, recent_bodies: int, max_body_chars: int, no_save: bool
) -> None:
"""Render a markdown debug report for one Claude Code session routed through the proxy
Examples:
lite debug claude
lite debug claude --session-id e96634a3-fa28-4083-b354-55542e2dca01
"""
resolved: Final = session_id or detect_claude_session_id(os.environ, CLAUDE_DIR)
if resolved is None:
raise click.ClickException(f"Could not find a Claude Code session. Pass --session-id or set ${SESSION_ID_ENV}.")
values: Final = cli_context_values(ctx)
base_url: Final = values["base_url"]
fetcher: Final = SpendLogsFetcher(HTTPClient(base_url, values["api_key"]))
outcome: Final = build_report(
fetcher=fetcher,
session_id=resolved,
base_url=base_url,
recent_bodies=recent_bodies,
max_chars=max_body_chars,
)
if isinstance(outcome, DebugFailure):
raise click.ClickException(outcome.message)
click.echo(outcome)
if not no_save:
path: Final = write_report(outcome, resolved, REPORT_DIR)
click.echo(f"Saved to {path}", err=True)
@debug.command("install-claude-command")
def debug_install_claude_command() -> None:
"""Install the /debug-lite slash command into ~/.claude/commands so Claude Code can run `lite debug claude`"""
path: Final = install_slash_command(CLAUDE_DIR)
click.echo(f"Installed /{SLASH_COMMAND_NAME}: {path}")
click.echo("Restart Claude Code (or start a new session), then type /debug-lite.")

View file

@ -14,6 +14,7 @@ from .commands.autoroute.commands import autoroute_group
from .commands.chat import chat
from .commands.config import config_commands, get_config_value, hidden_command_names
from .commands.credentials import credentials
from .commands.debug import debug
from .commands.encryption import encryption
from .commands.http import http
from .commands.keys import keys
@ -143,6 +144,7 @@ cli.add_command(encryption)
cli.add_command(chat)
# Add the http command group
cli.add_command(http)
cli.add_command(debug)
# Add the keys command group
cli.add_command(keys)
# Add the teams command group

View file

@ -0,0 +1,231 @@
import json
import os
import time
import pytest
import requests
import responses
from click.testing import CliRunner
from litellm.proxy.client.cli import cli
from litellm.proxy.client.cli.commands import debug as debug_module
from litellm.proxy.client.cli.commands.debug import (
SLASH_COMMAND_NAME,
detect_claude_session_id,
install_slash_command,
)
SESSION = "e96634a3-fa28-4083-b354-55542e2dca01"
OK_ROW = {
"request_id": "req-ok",
"startTime": "2026-09-02T10:00:00",
"endTime": "2026-09-02T10:00:02",
"model": "claude-opus-4-1",
"custom_llm_provider": "anthropic",
"status": "success",
"spend": 0.0125,
"prompt_tokens": 100,
"completion_tokens": 20,
"metadata": {"status": "success"},
}
FAILED_ROW = {
"request_id": "req-failed",
"startTime": "2026-09-02T10:01:00",
"endTime": "2026-09-02T10:01:01",
"model": "claude-opus-4-1",
"custom_llm_provider": "anthropic",
"status": "failure",
"spend": 0.0,
"prompt_tokens": 0,
"completion_tokens": 0,
"metadata": json.dumps(
{
"status": "failure",
"error_information": {
"error_code": "400",
"error_class": "BadRequestError",
"error_message": "`prompt` is required when `stop` is not true.",
},
}
),
}
PROXY = "http://localhost:4000"
def _mock_proxy(rows, payloads):
responses.get(
f"{PROXY}/spend/logs/session/ui",
json={"data": rows, "total": len(rows), "page": 1, "page_size": 100, "total_pages": 1},
match=[responses.matchers.query_param_matcher({"session_id": SESSION}, strict_match=False)],
)
for request_id, payload in payloads.items():
responses.get(f"{PROXY}/spend/logs/ui/{request_id}", json=payload)
def _called_paths():
return [c.request.path_url.split("?")[0] for c in responses.calls]
@pytest.fixture(autouse=True)
def env(monkeypatch, tmp_path):
monkeypatch.setenv("LITELLM_PROXY_URL", PROXY)
monkeypatch.setenv("LITELLM_PROXY_API_KEY", "sk-test")
monkeypatch.setattr(debug_module, "REPORT_DIR", tmp_path / "reports")
monkeypatch.setattr(debug_module, "CLAUDE_DIR", tmp_path / "claude")
@responses.activate
def test_report_includes_spend_error_and_bodies_for_failed_turn(tmp_path):
payloads = {
"req-failed": {
"proxy_server_request": {"body": {"model": "claude-opus-4-1", "messages": [{"role": "user"}]}},
"response": {"error": {"message": "`prompt` is required"}},
},
"req-ok": {"proxy_server_request": {"body": {"model": "claude-opus-4-1"}}, "response": {"id": "msg_1"}},
}
_mock_proxy([FAILED_ROW, OK_ROW], payloads)
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION, "--recent-bodies", "0"])
assert result.exit_code == 0, result.output
assert "turns: 2, failed: 1" in result.output
assert "total spend: $0.012500" in result.output
assert "### 1. ok claude-opus-4-1" in result.output
assert "### 2. FAILED claude-opus-4-1" in result.output
assert "`400` BadRequestError" in result.output
assert "`prompt` is required when `stop` is not true." in result.output
assert '"messages"' in result.output
assert "msg_1" not in result.output
assert _called_paths() == ["/spend/logs/session/ui", "/spend/logs/ui/req-failed"]
saved = tmp_path / "reports" / f"claude-{SESSION}.md"
assert result.stdout.startswith(saved.read_text())
assert "### 2. FAILED" in saved.read_text()
@responses.activate
def test_recent_bodies_fetches_latest_turns_even_when_successful():
_mock_proxy([OK_ROW], {"req-ok": {"proxy_server_request": {"body": {"x": 1}}, "response": {"id": "msg_1"}}})
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION, "--no-save"])
assert result.exit_code == 0, result.output
assert "msg_1" in result.output
assert _called_paths() == ["/spend/logs/session/ui", "/spend/logs/ui/req-ok"]
@responses.activate
def test_bodies_are_truncated_to_max_chars():
_mock_proxy([OK_ROW], {"req-ok": {"proxy_server_request": {"body": "a" * 5000}, "response": None}})
result = CliRunner().invoke(
cli, ["debug", "claude", "--session-id", SESSION, "--no-save", "--max-body-chars", "200"]
)
assert result.exit_code == 0, result.output
assert "truncated" in result.output
assert "a" * 300 not in result.output
@responses.activate
def test_no_rows_is_a_clear_error():
_mock_proxy([], {})
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION])
assert result.exit_code != 0
assert "No spend logs found for session" in result.output
def test_no_session_id_anywhere_is_a_clear_error(monkeypatch):
monkeypatch.delenv("CLAUDE_CODE_SESSION_ID", raising=False)
result = CliRunner().invoke(cli, ["debug", "claude"])
assert result.exit_code != 0
assert "Could not find a Claude Code session" in result.output
OLD_SESSION = "0f3c2b1a-1111-4222-8333-444455556666"
NEW_SESSION = "2d79c54d-4644-4708-b03e-95395ef9ecbd"
def test_detect_session_id_prefers_env_then_newest_session_transcript(tmp_path):
project = tmp_path / "projects" / "-Users-me-repo"
project.mkdir(parents=True)
old = project / f"{OLD_SESSION}.jsonl"
new = project / f"{NEW_SESSION}.jsonl"
subagent = project / "agent-a1b2c3d4.jsonl"
old.write_text("{}")
new.write_text("{}")
subagent.write_text("{}")
now = time.time()
os.utime(old, (now - 100, now - 100))
os.utime(new, (now - 50, now - 50))
os.utime(subagent, (now, now))
assert detect_claude_session_id({}, tmp_path) == NEW_SESSION
assert detect_claude_session_id({"CLAUDE_CODE_SESSION_ID": "from-env"}, tmp_path) == "from-env"
assert detect_claude_session_id({"CLAUDE_SESSION_ID": "stale-name"}, tmp_path) == NEW_SESSION
assert detect_claude_session_id({}, tmp_path / "missing") is None
def test_install_slash_command_writes_runnable_command_file(tmp_path):
path = install_slash_command(tmp_path)
assert path == tmp_path / "commands" / f"{SLASH_COMMAND_NAME}.md"
body = path.read_text()
assert body.startswith("---\n")
assert "allowed-tools: Bash(lite debug claude:*)" in body
assert "!`lite debug claude $ARGUMENTS`" in body
result = CliRunner().invoke(cli, ["debug", "install-claude-command"])
assert result.exit_code == 0, result.output
assert "/debug-lite" in result.output
@responses.activate
def test_rejected_key_is_a_clear_error_not_a_traceback():
responses.get(
f"{PROXY}/spend/logs/session/ui",
status=401,
json={"error": {"message": "Authentication Error, Invalid proxy server token passed", "code": "401"}},
)
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION])
assert isinstance(result.exception, SystemExit), result.exception
assert result.exit_code == 1
assert "401" in result.output
assert "Invalid proxy server token passed" in result.output
@responses.activate
def test_unreachable_proxy_is_a_clear_error_not_a_traceback():
responses.get(f"{PROXY}/spend/logs/session/ui", body=requests.ConnectionError("Connection refused"))
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION])
assert isinstance(result.exception, SystemExit), result.exception
assert result.exit_code == 1
assert "Connection refused" in result.output
@responses.activate
def test_non_json_proxy_response_is_a_clear_error_not_a_traceback():
responses.get(f"{PROXY}/spend/logs/session/ui", body="<html>502 Bad Gateway</html>")
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION])
assert isinstance(result.exception, SystemExit), result.exception
assert result.exit_code == 1
assert "/spend/logs/session/ui failed" in result.output
@responses.activate
def test_logged_content_with_code_fences_stays_inside_its_fence():
fenced_error_row = {
**FAILED_ROW,
"metadata": {
"status": "failure",
"error_information": {"error_code": "400", "error_message": "bad\n```\nrequest"},
},
}
_mock_proxy([fenced_error_row], {"req-failed": {"proxy_server_request": None, "response": "x\n````\ny"}})
result = CliRunner().invoke(cli, ["debug", "claude", "--session-id", SESSION, "--no-save"])
assert result.exit_code == 0, result.output
assert "````\nbad\n```\nrequest\n````\n" in result.output
assert "`````json\nx\n````\ny\n`````\n" in result.output