Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/cli/commands/debug.py: 0%
200 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""`lite debug claude`: one-shot debug report for a Claude Code session routed through the proxy.
3Claude Code puts its session id in `metadata.user_id`, which the proxy lifts into
4`LiteLLM_SpendLogs.session_id`. This command pulls every turn of that session, plus
5the request / response bodies for failures and the most recent turns, and renders a
6single markdown report that can be pasted into a bug report or handed to another agent.
7"""
9import json
10import os
11import re
12from collections.abc import Mapping, Sequence
13from dataclasses import dataclass
14from datetime import datetime, timezone
15from pathlib import Path
16from types import MappingProxyType
17from typing import Final
19import click
20import requests
21from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError, field_validator
23from ...http_client import HTTPClient
24from ._cli_context import cli_context_values
26CLAUDE_DIR: Final = Path.home() / ".claude"
27REPORT_DIR: Final = Path.home() / ".litellm" / "debug"
28SESSION_ID_ENV: Final = "CLAUDE_CODE_SESSION_ID"
29SLASH_COMMAND_NAME: Final = "debug-lite"
30SLASH_COMMAND_BODY: Final = """---
31description: Pull the LiteLLM debug report (spend, request, response, error) for this Claude Code session
32allowed-tools: Bash(lite debug claude:*)
33---
34Below is the LiteLLM debug report for this Claude Code session. Summarize the failing
35request(s) in a few sentences (model, error, request id) and tell me the path the full
36report was saved to so I can hand it off. If nothing failed, say so.
38!`lite debug claude $ARGUMENTS`
39"""
42@dataclass(frozen=True, slots=True)
43class DebugFailure:
44 message: str
47class ErrorInformation(BaseModel):
48 model_config = ConfigDict(frozen=True, extra="ignore")
50 error_code: str | None = None
51 error_class: str | None = None
52 error_message: str | None = None
53 llm_provider: str | None = None
56class SpendLogMetadata(BaseModel):
57 model_config = ConfigDict(frozen=True, extra="ignore")
59 status: str | None = None
60 error_information: ErrorInformation | None = None
63class SpendLogRow(BaseModel):
64 model_config = ConfigDict(frozen=True, extra="ignore", populate_by_name=True)
66 request_id: str
67 start_time: str | None = Field(default=None, alias="startTime")
68 end_time: str | None = Field(default=None, alias="endTime")
69 model: str | None = None
70 model_group: str | None = None
71 custom_llm_provider: str | None = None
72 api_base: str | None = None
73 call_type: str | None = None
74 status: str | None = None
75 spend: float = 0.0
76 prompt_tokens: int = 0
77 completion_tokens: int = 0
78 total_tokens: int = 0
79 metadata: SpendLogMetadata = SpendLogMetadata()
81 @field_validator("metadata", mode="before")
82 @classmethod
83 def _parse_metadata(cls, value: object) -> object:
84 if value is None:
85 return SpendLogMetadata()
86 if isinstance(value, str):
87 return json.loads(value) if value else SpendLogMetadata()
88 return value
90 @property
91 def failed(self) -> bool:
92 return (self.status or self.metadata.status) == "failure"
94 @property
95 def error(self) -> ErrorInformation | None:
96 return self.metadata.error_information
99class SessionLogsPage(BaseModel):
100 model_config = ConfigDict(frozen=True, extra="ignore")
102 data: tuple[SpendLogRow, ...]
103 total: int
104 total_pages: int
107class RequestResponsePayload(BaseModel):
108 model_config = ConfigDict(frozen=True, extra="ignore")
110 proxy_server_request: JsonValue = None
111 response: JsonValue = None
112 messages: JsonValue = None
115_SESSION_PAGE: Final = TypeAdapter(SessionLogsPage)
116_PAYLOAD: Final[TypeAdapter[RequestResponsePayload | None]] = TypeAdapter(RequestResponsePayload | None)
117_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
119_SESSION_PAGE_SIZE: Final = 100
120_TRANSPORT_BODY_CHARS: Final = 500
121_SESSION_TRANSCRIPT_STEM: Final = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
124def detect_claude_session_id(env: Mapping[str, str], claude_dir: Path) -> str | None:
125 explicit: Final = env.get(SESSION_ID_ENV)
126 if explicit:
127 return explicit
128 transcripts: Final = tuple(
129 path for path in claude_dir.glob("projects/*/*.jsonl") if _SESSION_TRANSCRIPT_STEM.fullmatch(path.stem)
130 )
131 if not transcripts:
132 return None
133 newest: Final = max(transcripts, key=lambda p: p.stat().st_mtime)
134 return newest.stem
137def _transport_failure(uri: str, error: requests.exceptions.RequestException) -> DebugFailure:
138 body: Final = error.response.text[:_TRANSPORT_BODY_CHARS] if error.response is not None else ""
139 detail: Final = f"\n{body}" if body else ""
140 return DebugFailure(f"GET {uri} failed: {error}{detail}")
143class SpendLogsFetcher:
144 def __init__(self, http: HTTPClient) -> None:
145 self._http = http
147 def session_rows(self, session_id: str) -> tuple[SpendLogRow, ...] | DebugFailure:
148 first: Final = self._page(session_id, 1)
149 if isinstance(first, DebugFailure):
150 return first
151 rest: Final = tuple(self._page(session_id, page) for page in range(2, first.total_pages + 1))
152 failed_page: Final = next((page for page in rest if isinstance(page, DebugFailure)), None)
153 if failed_page is not None:
154 return failed_page
155 rows: Final = first.data + tuple(row for page in rest if isinstance(page, SessionLogsPage) for row in page.data)
156 return tuple(sorted(rows, key=lambda r: r.start_time or ""))
158 def _get(self, uri: str, params: Mapping[str, str | int] | None = None) -> JsonValue | DebugFailure:
159 try:
160 return _JSON.validate_python(self._http.request("GET", uri, params=params)) # pyright: ignore[reportUnknownMemberType] # HTTPClient.request is untyped
161 except requests.exceptions.RequestException as e:
162 return _transport_failure(uri, e)
164 def _page(self, session_id: str, page: int) -> SessionLogsPage | DebugFailure:
165 uri: Final = "/spend/logs/session/ui"
166 raw: Final = self._get(
167 uri, MappingProxyType({"session_id": session_id, "page": page, "page_size": _SESSION_PAGE_SIZE})
168 )
169 if isinstance(raw, DebugFailure):
170 return raw
171 try:
172 return _SESSION_PAGE.validate_python(raw)
173 except ValidationError as e:
174 return DebugFailure(f"Unexpected {uri} response: {e}")
176 def payload(self, request_id: str) -> RequestResponsePayload | None | DebugFailure:
177 uri: Final = f"/spend/logs/ui/{request_id}"
178 raw: Final = self._get(uri)
179 if isinstance(raw, DebugFailure):
180 return raw
181 try:
182 return _PAYLOAD.validate_python(raw)
183 except ValidationError as e:
184 return DebugFailure(f"Unexpected {uri} response: {e}")
187def _fmt_json(value: JsonValue, max_chars: int) -> str:
188 text: Final = value if isinstance(value, str) else json.dumps(value, indent=2, default=str)
189 if len(text) <= max_chars:
190 return text
191 return f"{text[:max_chars]}\n... (truncated, {len(text) - max_chars} more chars)"
194def _fenced(text: str, info: str = "") -> tuple[str, str, str]:
195 longest_run: Final = max((len(run) for run in re.findall(r"`+", text)), default=0)
196 fence: Final = "`" * max(3, longest_run + 1)
197 return (f"{fence}{info}", text, fence)
200def _row_section(row: SpendLogRow, index: int, payload: RequestResponsePayload | None, max_chars: int) -> str:
201 err: Final = row.error
202 error_lines: Final = (
203 (
204 f"- error: `{err.error_code or '?'}` {err.error_class or ''}".rstrip(),
205 "",
206 *_fenced(err.error_message or ""),
207 )
208 if err is not None and row.failed
209 else ()
210 )
211 body_lines: Final = (
212 (
213 "",
214 "<details><summary>request body</summary>",
215 "",
216 *_fenced(_fmt_json(payload.proxy_server_request, max_chars), "json"),
217 "</details>",
218 "",
219 "<details><summary>response</summary>",
220 "",
221 *_fenced(_fmt_json(payload.response, max_chars), "json"),
222 "</details>",
223 )
224 if payload is not None
225 else ()
226 )
227 header: Final = f"### {index}. {'FAILED' if row.failed else 'ok'} {row.model or row.model_group or '?'}"
228 facts: Final = (
229 f"- request_id: `{row.request_id}`",
230 f"- time: {row.start_time} -> {row.end_time}",
231 f"- provider: {row.custom_llm_provider or '?'} ({row.api_base or 'n/a'}), call_type: {row.call_type or '?'}",
232 f"- spend: ${row.spend:.6f}, tokens: {row.prompt_tokens} in / {row.completion_tokens} out",
233 )
234 return "\n".join((header, *facts, *error_lines, *body_lines))
237def render_report(
238 *,
239 session_id: str,
240 base_url: str,
241 rows: Sequence[SpendLogRow],
242 payloads: Mapping[str, RequestResponsePayload | None],
243 max_chars: int,
244) -> str:
245 failures: Final = tuple(r for r in rows if r.failed)
246 summary: Final = (
247 f"# LiteLLM debug report: Claude Code session `{session_id}`",
248 "",
249 f"- proxy: {base_url}",
250 f"- generated: {datetime.now(timezone.utc).isoformat(timespec='seconds')}",
251 f"- turns: {len(rows)}, failed: {len(failures)}",
252 f"- total spend: ${sum(r.spend for r in rows):.6f}",
253 f"- models: {', '.join(sorted(frozenset(r.model or r.model_group or '?' for r in rows))) or 'n/a'}",
254 "",
255 "Bodies are included for failed turns and the most recent turns. "
256 "Bodies are empty unless the proxy runs with `general_settings.store_prompts_in_spend_logs: true`.",
257 "",
258 "## Turns",
259 "",
260 )
261 sections: Final = tuple(
262 _row_section(row, i, payloads.get(row.request_id), max_chars) for i, row in enumerate(rows, start=1)
263 )
264 return "\n".join(summary) + "\n\n".join(sections) + "\n"
267def build_report(
268 *,
269 fetcher: SpendLogsFetcher,
270 session_id: str,
271 base_url: str,
272 recent_bodies: int,
273 max_chars: int,
274) -> str | DebugFailure:
275 rows: Final = fetcher.session_rows(session_id)
276 if isinstance(rows, DebugFailure):
277 return rows
278 if not rows:
279 return DebugFailure(
280 f"No spend logs found for session {session_id!r} on {base_url}. "
281 "Is Claude Code routed through this proxy (`lite up`), and does your key have log access?"
282 )
283 wanted: Final = frozenset(r.request_id for r in rows if r.failed) | frozenset(
284 r.request_id for r in rows[-recent_bodies:] if recent_bodies > 0
285 )
286 fetched: Final = MappingProxyType({rid: fetcher.payload(rid) for rid in sorted(wanted)})
287 failed_payload: Final = next((p for p in fetched.values() if isinstance(p, DebugFailure)), None)
288 if failed_payload is not None:
289 return failed_payload
290 payloads: Final = MappingProxyType({rid: p for rid, p in fetched.items() if not isinstance(p, DebugFailure)})
291 return render_report(session_id=session_id, base_url=base_url, rows=rows, payloads=payloads, max_chars=max_chars)
294def write_report(report: str, session_id: str, report_dir: Path) -> Path:
295 report_dir.mkdir(parents=True, exist_ok=True)
296 path: Final = report_dir / f"claude-{session_id}.md"
297 path.write_text(report, encoding="utf-8")
298 path.chmod(0o600)
299 return path
302def install_slash_command(claude_dir: Path) -> Path:
303 commands_dir: Final = claude_dir / "commands"
304 commands_dir.mkdir(parents=True, exist_ok=True)
305 path: Final = commands_dir / f"{SLASH_COMMAND_NAME}.md"
306 path.write_text(SLASH_COMMAND_BODY, encoding="utf-8")
307 return path
310@click.group()
311def debug() -> None:
312 """Pull debug reports (spend, request, response, error) for coding-agent sessions"""
315@debug.command("claude")
316@click.option(
317 "--session-id",
318 default=None,
319 help=f"Claude Code session id. Defaults to ${SESSION_ID_ENV}, else the most recently used transcript in ~/.claude",
320)
321@click.option(
322 "--recent-bodies",
323 default=3,
324 show_default=True,
325 type=click.IntRange(min=0),
326 help="Also include request/response bodies for the N most recent turns (failed turns always get bodies)",
327)
328@click.option(
329 "--max-body-chars",
330 default=20_000,
331 show_default=True,
332 type=click.IntRange(min=100),
333 help="Truncate each request/response body to this many characters",
334)
335@click.option("--no-save", is_flag=True, help="Print only, do not write the report under ~/.litellm/debug")
336@click.pass_context
337def debug_claude(
338 ctx: click.Context, session_id: str | None, recent_bodies: int, max_body_chars: int, no_save: bool
339) -> None:
340 """Render a markdown debug report for one Claude Code session routed through the proxy
342 Examples:
343 lite debug claude
344 lite debug claude --session-id e96634a3-fa28-4083-b354-55542e2dca01
345 """
346 resolved: Final = session_id or detect_claude_session_id(os.environ, CLAUDE_DIR)
347 if resolved is None:
348 raise click.ClickException(f"Could not find a Claude Code session. Pass --session-id or set ${SESSION_ID_ENV}.")
349 values: Final = cli_context_values(ctx)
350 base_url: Final = values["base_url"]
351 fetcher: Final = SpendLogsFetcher(HTTPClient(base_url, values["api_key"]))
352 outcome: Final = build_report(
353 fetcher=fetcher,
354 session_id=resolved,
355 base_url=base_url,
356 recent_bodies=recent_bodies,
357 max_chars=max_body_chars,
358 )
359 if isinstance(outcome, DebugFailure):
360 raise click.ClickException(outcome.message)
361 click.echo(outcome)
362 if not no_save:
363 path: Final = write_report(outcome, resolved, REPORT_DIR)
364 click.echo(f"Saved to {path}", err=True)
367@debug.command("install-claude-command")
368def debug_install_claude_command() -> None:
369 """Install the /debug-lite slash command into ~/.claude/commands so Claude Code can run `lite debug claude`"""
370 path: Final = install_slash_command(CLAUDE_DIR)
371 click.echo(f"Installed /{SLASH_COMMAND_NAME}: {path}")
372 click.echo("Restart Claude Code (or start a new session), then type /debug-lite.")