Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/cli/commands/statusline

1"""Claude Code status line and Codex Stop hook for auto-routed sessions.

3`lite` copies this file to ~/.litellm/statusline.py with a CLI version header when known and registers

4it as Claude Code's `statusLine` command and as Codex's `[[hooks.Stop]]` command, so it must stay

5standard-library only and must never import litellm. Claude Code re-runs it on every

6status refresh (about every 300ms while typing), so the proxy is asked at most once per

7TTL per session and every other refresh is served from a small on-disk cache that holds

8only the proxy's answer, never the key.

10Claude Code pipes a JSON payload on stdin (session_id, transcript_path, model). After the

11first foreground assistant response, the routed model comes from the proxy's session

12record, falling back to the latest foreground assistant `message.model` in the transcript

13when no record is available. Codex pipes its Stop event instead (hook_event_name, session_id)

14and prints the session record as a `systemMessage` for the transcript. The proxy key is read

15from the agent's own environment (the static token `lite configure claude` writes); nothing

16here spawns a credential helper.

18The routed model and cost figures come from GET /auto_router/session on the proxy, which

19reads the per-session rollup written by the asynchronous spend flush. The record and cache

20can briefly lag a completed turn.

21"""

23from __future__ import annotations

25import hashlib

26import json

27import os

28import sys

29import tempfile

30import time

31import unicodedata

32import urllib.error

33import urllib.request

34from collections.abc import Callable, Mapping

35from math import isfinite

36from pathlib import Path

37from types import MappingProxyType

38from typing import IO, Final, NamedTuple, Protocol

39from urllib.parse import urlencode

41SESSION_ENDPOINT: Final = "/auto_router/session"

42CACHE_TTL_SECONDS: Final = 5.0

43FETCH_TIMEOUT_SECONDS: Final = 3

44BAR_WIDTH: Final = 24

45BAR_FULL: Final = "\u2588"

46BAR_EMPTY: Final = "\u2591"

47SEPARATOR: Final = " \u00b7 "

48TRANSCRIPT_SCAN_LIMIT_BYTES: Final = 4 * 1024 * 1024

49CLAUDE_BASE_URL_ENV_KEYS: Final = ("ANTHROPIC_BASE_URL",)

50CLAUDE_API_KEY_ENV_KEYS: Final = ("ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_API_KEY")

51CODEX_BASE_URL_ENV_KEYS: Final = ("OPENAI_BASE_URL",)

52CODEX_API_KEY_ENV_KEYS: Final = ("OPENAI_API_KEY",)

53CODEX_STOP_EVENT: Final = "Stop"

54SYNTHETIC_MODEL: Final = "<synthetic>"

55RESET: Final = "\033[0m"

56BOLD: Final = "\033[1m"

57DIM: Final = "\033[90m"

58LITELLM_COLOR: Final = "\033[38;2;79;70;229m"

59BASELINE_COLOR: Final = "\033[38;2;217;119;87m"

60EMPTY: Final[Mapping[str, object]] = MappingProxyType({})

61EMPTY_ENV: Final[Mapping[str, str]] = MappingProxyType({})

64class Session(NamedTuple):

65 router_name: str

66 last_model: str

67 spend: float

68 baseline_spend: float | None

69 baseline_model: str | None

70 turns: int | None = None

71 savings_estimated_turns: int | None = None

72 savings_estimated_actual_spend: float | None = None

75class Credentials(NamedTuple):

76 base_url: str

77 api_key: str

79 @property

80 def usable(self) -> bool:

81 return bool(self.base_url and self.api_key)

84class Fetched(NamedTuple):

85 session: Session | None

86 definitive: bool

89class Fetch(Protocol):

90 def __call__(self, credentials: Credentials, session_id: str) -> Fetched: ...

93def as_mapping(value: object) -> Mapping[str, object]:

94 return value if isinstance(value, dict) else EMPTY

97def as_str(value: object) -> str:

98 return value if isinstance(value, str) else ""

100

101def printable(value: object) -> str:

102 """Labels come from the transcript, the proxy, and Claude Code's model cache, none of which this script

103 controls, and every one is written to a terminal: a control character (ESC, BEL, C1) in a model name

104 could redraw the screen or set the clipboard, so only printable text survives."""

105 return "".join(character for character in as_str(value) if character.isprintable())

106

107

108def load_json(raw: bytes | str) -> object:

109 try:

110 return json.loads(raw)

111 except ValueError:

112 return None

113

114

115def resolve_base_url(env: Mapping[str, str], keys: tuple[str, ...]) -> str:

116 raw: Final = next((env[key] for key in keys if env.get(key)), "").strip().rstrip("/")

117 return raw.removesuffix("/v1")

118

119

120def resolve_api_key(env: Mapping[str, str], keys: tuple[str, ...]) -> str:

121 return next((env[key] for key in keys if env.get(key)), "").strip()

122

123

124def claude_credentials(env: Mapping[str, str]) -> Credentials:

125 """Claude Code's own resolution order, so the key-scoped lookup runs as the principal that wrote the rows:

126 ANTHROPIC_AUTH_TOKEN, then ANTHROPIC_API_KEY. A `lite` variable such as LITELLM_PROXY_API_KEY is not a

127 key Claude Code ever sends, so honoring it would ask as someone else. An apiKeyHelper is never run: a

128 status line refreshes every few hundred milliseconds, and spawning a credential helper that often is

129 how a keychain prompt ends up on screen a hundred times."""

130 return Credentials(resolve_base_url(env, CLAUDE_BASE_URL_ENV_KEYS), resolve_api_key(env, CLAUDE_API_KEY_ENV_KEYS))

131

132

133def codex_credentials(env: Mapping[str, str]) -> Credentials:

134 return Credentials(resolve_base_url(env, CODEX_BASE_URL_ENV_KEYS), resolve_api_key(env, CODEX_API_KEY_ENV_KEYS))

135

136

137def _transcript_line_model(line: bytes) -> str:

138 """A `<synthetic>` model is Claude Code's own marker for a locally produced message (an API error, a

139 resume note), not a served model, so it is skipped like a sidechain line."""

140 item: Final = as_mapping(load_json(line))

141 if item.get("type") != "assistant" or item.get("isSidechain") is True or item.get("agentId"):

142 return ""

143 model: Final = printable(as_mapping(item.get("message")).get("model"))

144 return "" if model == SYNTHETIC_MODEL else model

145

146

147def latest_transcript_model(transcript_path: str) -> str:

148 if not transcript_path:

149 return ""

150 try:

151 with Path(transcript_path).open("rb") as transcript:

152 size: Final = transcript.seek(0, os.SEEK_END)

153 transcript.seek(max(0, size - TRANSCRIPT_SCAN_LIMIT_BYTES))

154 tail: Final = transcript.read()

155 except OSError:

156 return ""

157 return next((model for line in reversed(tail.split(b"\n")) if (model := _transcript_line_model(line))), "")

158

159

160def model_label(model: str, config_dir: Path) -> str:

161 bare: Final = model.rsplit("/", 1)[-1]

162 try:

163 raw: Final = (config_dir / "cache" / "gateway-models.json").read_bytes()

164 except OSError:

165 return bare

166 listed: Final = as_mapping(load_json(raw)).get("models")

167 if not isinstance(listed, list):

168 return bare

169 entries: Final = tuple(as_mapping(entry) for entry in listed)

170 return next(

171 (

172 printable(entry.get("display_name"))

173 for entry in entries

174 if entry.get("id") in (model, bare) and printable(entry.get("display_name"))

175 ),

176 bare,

177 )

178

179

180def baseline_label(model: str, config_dir: Path) -> str:

181 labelled: Final = model_label(model, config_dir)

182 if labelled != model.rsplit("/", 1)[-1]:

183 return labelled

184 return " ".join(word.capitalize() for word in labelled.replace("-", " ").split())

185

186

187def fetch_session(credentials: Credentials, session_id: str) -> Fetched:

188 """Any 4xx is this credential's definite answer (no row, no access, expired login) and is cached for the

189 TTL; a 5xx or transport failure is not, so the next refresh tries again."""

190 query: Final = urlencode((("session_id", session_id),))

191 request: Final = urllib.request.Request(

192 f"{credentials.base_url}{SESSION_ENDPOINT}?{query}",

193 headers={ # mutable-ok: urllib.request.Request takes a dict

194 "Authorization": f"Bearer {credentials.api_key}",

195 "Accept": "application/json",

196 },

197 )

198 try:

199 with urllib.request.urlopen(request, timeout=FETCH_TIMEOUT_SECONDS) as response:

200 raw: Final[bytes] = response.read()

201 except urllib.error.HTTPError as error:

202 return Fetched(session=None, definitive=400 <= error.code < 500)

203 except (urllib.error.URLError, OSError):

204 return Fetched(session=None, definitive=False)

205 session: Final = _session_from_payload(as_mapping(load_json(raw)))

206 return Fetched(session=session, definitive=session is not None)

207

208

209def _session_from_payload(payload: Mapping[str, object]) -> Session | None:

210 router_name: Final = printable(payload.get("router_name"))

211 last_model: Final = printable(payload.get("last_model"))

212 spend: Final = payload.get("spend")

213 baseline_spend: Final = payload.get("savings_estimated_baseline_spend", payload.get("baseline_spend"))

214 turns: Final = payload.get("turns")

215 estimated_turns: Final = payload.get("savings_estimated_turns")

216 estimated_actual: Final = payload.get("savings_estimated_actual_spend")

217 if not router_name or not last_model:

218 return None

219 if not isinstance(spend, (int, float)) or isinstance(spend, bool) or not isfinite(spend):

220 return None

221 if baseline_spend is not None and (

222 not isinstance(baseline_spend, (int, float)) or isinstance(baseline_spend, bool) or not isfinite(baseline_spend)

223 ):

224 return None

225 return Session(

226 router_name=router_name,

227 last_model=last_model,

228 spend=float(spend),

229 baseline_spend=float(baseline_spend) if baseline_spend is not None else None,

230 baseline_model=printable(payload.get("baseline_model")) or None,

231 turns=turns if isinstance(turns, int) and not isinstance(turns, bool) and turns >= 0 else None,

232 savings_estimated_turns=(

233 estimated_turns

234 if isinstance(estimated_turns, int) and not isinstance(estimated_turns, bool) and estimated_turns >= 0

235 else (0 if estimated_turns is not None else None)

236 ),

237 savings_estimated_actual_spend=(

238 float(estimated_actual)

239 if isinstance(estimated_actual, (int, float))

240 and not isinstance(estimated_actual, bool)

241 and isfinite(estimated_actual)

242 and estimated_actual >= 0

243 else None

244 ),

245 )

246

247

248def cache_path(cache_dir: Path, credentials: Credentials, session_id: str) -> Path:

249 identity: Final = "\n".join((credentials.base_url, credentials.api_key, session_id))

250 return cache_dir / hashlib.sha256(identity.encode()).hexdigest()

251

252

253def load_session(

254 credentials: Credentials,

255 session_id: str,

256 cache_dir: Path,

257 fetch: Fetch = fetch_session,

258 now: Callable[[], float] = time.time,

259) -> Session | None:

260 path: Final = cache_path(cache_dir, credentials, session_id)

261 cached: Final = _read_cache(path)

262 fetched_at: Final = cached.get("fetched_at")

263 if isinstance(fetched_at, (int, float)) and now() - fetched_at < CACHE_TTL_SECONDS:

264 return _session_from_payload(as_mapping(cached.get("session")))

265 fetched: Final = fetch(credentials, session_id)

266 if fetched.definitive:

267 _write_cache(path, fetched.session, now())

268 return fetched.session

269

270

271NOFOLLOW: Final = getattr(os, "O_NOFOLLOW", 0)

272

273

274def cache_dir_name() -> str:

275 return f"litellm-statusline-{os.getuid()}" if hasattr(os, "getuid") else "litellm-statusline"

276

277

278def _own_private_dir(directory: Path) -> bool:

279 """A shared temp root lets another local user pre-create the directory, so it must be ours and private

280 before anything is read or written under it. Windows has no uids or POSIX mode bits and a per-user temp

281 directory already, so there it only has to exist and not be a link."""

282 try:

283 directory.mkdir(mode=0o700, parents=True, exist_ok=True)

284 status: Final = directory.lstat()

285 except OSError:

286 return False

287 if not os.path.isdir(directory) or os.path.islink(directory):

288 return False

289 if not hasattr(os, "getuid"):

290 return True

291 return status.st_uid == os.getuid() and not status.st_mode & 0o077

292

293

294def _read_cache(path: Path) -> Mapping[str, object]:

295 if not _own_private_dir(path.parent):

296 return EMPTY

297 try:

298 descriptor: Final = os.open(path, os.O_RDONLY | NOFOLLOW)

299 with os.fdopen(descriptor, "rb") as handle:

300 return as_mapping(load_json(handle.read()))

301 except OSError:

302 return EMPTY

303

304

305def _write_cache(path: Path, session: Session | None, fetched_at: float) -> None:

306 """Staged beside the entry and renamed into place, so a refresh reading the entry never sees a torn write."""

307 entry: Final = session._asdict() if session else None

308 body: Final = json.dumps({"fetched_at": fetched_at, "session": entry}) # mutable-ok: json.dumps takes a dict

309 if not _own_private_dir(path.parent):

310 return

311 try:

312 descriptor, staged = tempfile.mkstemp(dir=path.parent, prefix=".tmp-")

313 except OSError:

314 return

315 try:

316 with os.fdopen(descriptor, "w") as handle:

317 handle.write(body)

318 os.replace(staged, path)

319 except OSError:

320 Path(staged).unlink(missing_ok=True)

321

322

323def _bar(fraction: float, color: str, width: int, use_color: bool) -> str:

324 filled: Final = round(max(0.0, min(1.0, fraction)) * width)

325 if not use_color:

326 return BAR_FULL * filled + BAR_EMPTY * (width - filled)

327 return f"{color}{BAR_FULL * filled}{DIM}{BAR_EMPTY * (width - filled)}{RESET}"

328

329

330def _display_width(label: str) -> int:

331 return sum(

332 2 if unicodedata.east_asian_width(character) in ("W", "F") else 1

333 for character in label

334 if unicodedata.category(character) not in ("Mn", "Me")

335 )

336

337

338def render(model: str, session: Session | None, config_dir: Path, use_color: bool, bar_width: int = BAR_WIDTH) -> str:

339 def paint(code: str, text: str) -> str:

340 return f"{code}{text}{RESET}" if use_color else text

341

342 routed: Final = paint(BOLD, f"Routed to: {model}")

343 if session is None:

344 return routed

345 if session.savings_estimated_turns == 0 or session.baseline_spend is None:

346 return f"{routed}{SEPARATOR}Savings unavailable"

347 if session.baseline_model is None or session.baseline_spend <= 0:

348 return routed

349 if session.savings_estimated_turns is not None and (

350 session.savings_estimated_actual_spend is None

351 or session.turns is None

352 or session.savings_estimated_turns > session.turns

353 ):

354 return f"{routed}{SEPARATOR}Savings unavailable"

355 compared_spend: Final = (

356 session.savings_estimated_actual_spend

357 if session.savings_estimated_turns is not None and session.savings_estimated_actual_spend is not None

358 else session.spend

359 )

360 coverage: Final = (

361 f"{SEPARATOR}{session.savings_estimated_turns} of {session.turns} turns estimated"

362 if session.savings_estimated_turns is not None

363 else ""

364 )

365 reference: Final = baseline_label(session.baseline_model, config_dir)

366 pct: Final = round((session.baseline_spend - compared_spend) / session.baseline_spend * 100)

367 sign: Final = "-" if pct > 0 else "+" if pct < 0 else ""

368 delta: Final = paint(LITELLM_COLOR, f"{sign}{abs(pct)}% vs {reference}")

369 peak: Final = max(compared_spend, session.baseline_spend)

370 label_width: Final = max(_display_width(session.router_name), _display_width(reference))

371 rows: Final = (

372 (session.router_name, compared_spend, LITELLM_COLOR),

373 (reference, session.baseline_spend, BASELINE_COLOR),

374 )

375 lines: Final = (

376 f"{paint(DIM, label + ' ' * (label_width - _display_width(label)))} "

377 f"{_bar(amount / peak, color, bar_width, use_color)} "

378 f"{paint(DIM, f'${amount:.2f}')}"

379 for label, amount, color in rows

380 )

381 return "\n".join((f"{routed} {delta}{coverage}", *lines))

382

383

384def color_enabled(env: Mapping[str, str]) -> bool:

385 return env.get("NO_COLOR") is None and env.get("TERM", "") not in ("", "dumb")

386

387

388def status_line(

389 payload: Mapping[str, object], env: Mapping[str, str], config_dir: Path, cache_dir: Path, fetch: Fetch

390) -> str:

391 fallback: Final = printable(as_mapping(payload.get("model")).get("display_name"))

392 served: Final = latest_transcript_model(as_str(payload.get("transcript_path")))

393 if not served:

394 return fallback or "claude"

395 label: Final = model_label(served, config_dir)

396 session_id: Final = as_str(payload.get("session_id"))

397 credentials: Final = claude_credentials(env)

398 if not session_id or not credentials.usable:

399 return render(label, None, config_dir, color_enabled(env))

400 session: Final = load_session(credentials, session_id, cache_dir, fetch)

401 routed_label: Final = model_label(session.last_model, config_dir) if session is not None else label

402 return render(routed_label, session, config_dir, color_enabled(env))

403

404

405def codex_stop_message(

406 payload: Mapping[str, object], env: Mapping[str, str], config_dir: Path, cache_dir: Path, fetch: Fetch

407) -> str:

408 """No cache here: the Stop hook runs once per turn, and a first turn's cached absence would hide the record

409 the next turn finds."""

410 session_id: Final = as_str(payload.get("session_id"))

411 credentials: Final = codex_credentials(env)

412 if not session_id or not credentials.usable:

413 return ""

414 session: Final = fetch(credentials, session_id).session

415 if session is None:

416 return ""

417 text: Final = render(model_label(session.last_model, config_dir), session, config_dir, use_color=False)

418 return json.dumps({"systemMessage": f"\n{text}"}) # mutable-ok: json.dumps takes a dict

419

420

421def run(stdin: IO[str], stdout: IO[str], env: Mapping[str, str], fetch: Fetch = fetch_session) -> None:

422 """A failure renders each mode's own quiet fallback: Claude Code gets the label it already knows, Codex gets

423 nothing at all rather than a bare string it would reject as hook JSON."""

424 body: Final = as_mapping(load_json(stdin.read()))

425 codex: Final = body.get("hook_event_name") == CODEX_STOP_EVENT

426 config_dir: Final = Path(env.get("CLAUDE_CONFIG_DIR") or Path.home() / ".claude")

427 cache_dir: Final = (

428 Path(env.get("TMPDIR") or env.get("TEMP") or env.get("TMP") or tempfile.gettempdir()) / cache_dir_name()

429 )

430 try:

431 text: Final = (

432 codex_stop_message(body, env, config_dir, cache_dir, fetch)

433 if codex

434 else status_line(body, env, config_dir, cache_dir, fetch)

435 )

436 except Exception: # noqa: BLE001 # a status line must never break the agent session

437 stdout.write("" if codex else printable(as_mapping(body.get("model")).get("display_name")) or "claude")

438 return

439 stdout.write(text)

440

441

442if __name__ == "__main__":

443 run(sys.stdin, sys.stdout, os.environ)

Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/cli/commands/statusline_script.py: 0%

256 statements