Coverage for astrocyte/providers/claude_cli.py: 91%
202 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Claude Code CLI LLMProvider adapter — completion via the local ``claude`` binary.
3Runs ``claude -p`` (print mode) as a subprocess for every ``complete()`` call.
4Authentication is the CLI's own logged-in subscription: no ANTHROPIC_API_KEY is
5required, and any key present in the environment is deliberately NOT passed to
6the subprocess, so calls bill the subscription rather than an API key.
8Usage programmatically (the bench harness path)::
10 from astrocyte.providers.claude_cli import ClaudeCliProvider
12 llm = ClaudeCliProvider(model="haiku")
14Limitations (inherent to the CLI surface, documented rather than hidden):
16- ``temperature`` and ``max_tokens`` are accepted for protocol compatibility
17 but IGNORED — the CLI exposes no sampling controls. Results are therefore a
18 separate ablation track, never comparable to API-provider runs.
19- ``tools`` / native function calling is NOT supported and raises
20 ``NotImplementedError`` (loud failure beats silently dropping tool calls).
21- ``response_format`` (json_object / json_schema) is EMULATED by appending a
22 JSON-only instruction to the prompt. Downstream Astrocyte stages already
23 parse with ``_json_tolerant``, so prose slippage degrades gracefully.
24- ``embed()`` is NOT supported — the CLI has no embeddings surface. Compose
25 with :class:`~astrocyte.providers.local_embeddings.LocalEmbeddingsProvider`
26 via :class:`~astrocyte.providers.composite.CompositeLLMProvider`.
27- Each call pays ~1-3s of process startup on top of inference. Concurrency is
28 bounded by a semaphore (``ASTROCYTE_CLAUDE_CLI_MAX_CONCURRENCY``, default 4).
30Resilience (added after the first claude-native n=90 run lost ~1300 ingest
31calls to silent throttling and produced a bench over empty memories):
33- **Both streams captured on failure.** The CLI reports throttling on
34 stdout; logging stderr alone made every failure look empty.
35- **Rate-limit-aware backoff.** Output matching a throttle signature backs
36 off on a minutes scale (``ASTROCYTE_CLAUDE_CLI_RATE_LIMIT_BACKOFF``,
37 default 60s, doubling per attempt, capped at 600s) rather than the 2s/4s
38 ladder used for ordinary errors.
39- **Circuit breaker.** After ``ASTROCYTE_CLAUDE_CLI_BREAKER_THRESHOLD``
40 consecutive failures (default 5) *all* callers pause for
41 ``ASTROCYTE_CLAUDE_CLI_BREAKER_COOLDOWN`` seconds (default 300, doubling
42 up to 30min) — the pipeline stops rather than completing a run whose
43 memories were never populated.
44"""
46from __future__ import annotations
48import asyncio
49import json
50import logging
51import os
52import shutil
53import tempfile
54import time
55from typing import ClassVar
57from astrocyte.types import (
58 Completion,
59 ContentPart,
60 LLMCapabilities,
61 Message,
62 ToolDefinition,
63)
65logger = logging.getLogger("astrocyte.providers.claude_cli")
68def _flatten_content(content: str | list[ContentPart]) -> str:
69 """Render Message.content to plain text. Non-text parts are dropped
70 with a log line — the CLI is a text surface."""
71 if isinstance(content, str):
72 return content
73 parts: list[str] = []
74 for part in content:
75 text = getattr(part, "text", None)
76 if text:
77 parts.append(text)
78 else:
79 logger.debug("claude_cli: dropping non-text content part %r", type(part))
80 return "\n".join(parts)
83def _render_prompt(messages: list[Message], response_format: dict | None) -> str:
84 """Flatten a chat transcript into a single print-mode prompt.
86 System messages float to the top (in order); user/assistant turns are
87 role-labelled only when there is more than one non-system message, so
88 the common single-user-message case stays a clean bare prompt.
89 """
90 system_parts = [_flatten_content(m.content) for m in messages if m.role == "system"]
91 convo = [m for m in messages if m.role != "system"]
93 if len(convo) <= 1:
94 body = _flatten_content(convo[0].content) if convo else ""
95 else:
96 body = "\n\n".join(
97 f"[{m.role}]\n{_flatten_content(m.content)}" for m in convo
98 )
100 blocks: list[str] = []
101 if system_parts:
102 blocks.append("\n\n".join(p for p in system_parts if p))
103 if body:
104 blocks.append(body)
106 if response_format is not None:
107 rf_type = response_format.get("type") if isinstance(response_format, dict) else None
108 if rf_type == "json_schema":
109 schema = response_format.get("json_schema", {})
110 blocks.append(
111 "IMPORTANT: Respond with a single valid JSON object conforming "
112 "to this JSON schema. No prose, no code fences.\n"
113 f"{json.dumps(schema, ensure_ascii=False)}"
114 )
115 else: # json_object and any future variants degrade to the generic rule
116 blocks.append(
117 "IMPORTANT: Respond with a single valid JSON object only. "
118 "No prose, no code fences."
119 )
121 return "\n\n---\n\n".join(blocks)
124def _normalize_json_output(text: str) -> str | None:
125 """Best-effort normalization of model output to a bare JSON value.
127 Emulated JSON mode (see module docstring) means the model can wrap the
128 object in ```json fences or surrounding prose. Returns the canonical
129 JSON string when a parseable value can be recovered, else ``None``.
131 Recovery ladder (first hit wins):
132 1. The text parses as-is.
133 2. Code-fence stripped (```json ... ``` or bare ``` fences).
134 3. Outermost ``{...}`` or ``[...]`` slice of the raw text.
135 """
136 stripped = text.strip()
137 if not stripped:
138 return None
139 try:
140 json.loads(stripped)
141 return stripped
142 except json.JSONDecodeError:
143 pass
145 if stripped.startswith("```"):
146 newline = stripped.find("\n")
147 inner = stripped[newline + 1 :] if newline != -1 else stripped[3:]
148 inner = inner.rstrip()
149 if inner.endswith("```"):
150 inner = inner[:-3].rstrip()
151 try:
152 json.loads(inner)
153 return inner
154 except json.JSONDecodeError:
155 stripped = inner # fall through to brace slicing on the unfenced body
157 for open_ch, close_ch in (("{", "}"), ("[", "]")):
158 start = stripped.find(open_ch)
159 end = stripped.rfind(close_ch)
160 if start != -1 and end > start:
161 candidate = stripped[start : end + 1]
162 try:
163 json.loads(candidate)
164 return candidate
165 except json.JSONDecodeError:
166 continue
167 return None
170#: Substrings that indicate the CLI was refused for quota/rate reasons rather
171#: than a genuine error. Matched case-insensitively against combined
172#: stdout+stderr. The CLI reports these on stdout, which is why the first
173#: claude-native bench run saw 1329 failures with empty error text.
174_RATE_LIMIT_SIGNATURES: tuple[str, ...] = (
175 "rate limit",
176 "rate-limit",
177 "usage limit",
178 "limit reached",
179 "quota",
180 "too many requests",
181 "429",
182 "overloaded",
183 "try again later",
184 "please wait",
185 "capacity",
186)
189def _looks_rate_limited(text: str) -> bool:
190 """True when CLI output smells like throttling rather than a hard error."""
191 low = text.lower()
192 return any(sig in low for sig in _RATE_LIMIT_SIGNATURES)
195class _CircuitBreaker:
196 """Pauses *all* calls after a sustained failure burst.
198 Motivation: the first claude-native n=90 run ground through ~1300 failed
199 ingest calls and produced questions whose memories were empty — a bench
200 that reports a number but measures nothing. A burst of failures means the
201 upstream is refusing us; the correct response is to stop and wait, not to
202 keep spending quota on calls that will fail and silently degrade the data.
204 Opens after ``threshold`` consecutive failures, holds every caller for
205 ``cooldown`` seconds, then lets a single probe through. A successful call
206 closes it; a failure re-opens it with the cooldown doubled (capped).
207 """
209 def __init__(self, *, threshold: int = 5, cooldown: float = 300.0, max_cooldown: float = 1800.0) -> None:
210 self._threshold = threshold
211 self._base_cooldown = cooldown
212 self._cooldown = cooldown
213 self._max_cooldown = max_cooldown
214 self._consecutive = 0
215 self._open_until = 0.0
216 self._lock = asyncio.Lock()
218 async def await_closed(self) -> None:
219 """Block until the breaker is closed. No-op in the healthy path."""
220 while True:
221 async with self._lock:
222 remaining = self._open_until - time.monotonic()
223 if remaining <= 0:
224 return
225 logger.warning(
226 "claude_cli: circuit OPEN — pausing %.0fs before retrying "
227 "(sustained failures suggest rate limiting)", remaining,
228 )
229 await asyncio.sleep(min(remaining, 30.0))
231 async def record_success(self) -> None:
232 async with self._lock:
233 if self._consecutive or self._cooldown != self._base_cooldown:
234 logger.info("claude_cli: circuit CLOSED — calls succeeding again")
235 self._consecutive = 0
236 self._cooldown = self._base_cooldown
237 self._open_until = 0.0
239 async def record_failure(self) -> None:
240 async with self._lock:
241 self._consecutive += 1
242 if self._consecutive < self._threshold:
243 return
244 self._open_until = time.monotonic() + self._cooldown
245 logger.error(
246 "claude_cli: %d consecutive failures — opening circuit for %.0fs. "
247 "Ingest is PAUSED rather than proceeding with degraded memories.",
248 self._consecutive, self._cooldown,
249 )
250 self._cooldown = min(self._cooldown * 2, self._max_cooldown)
251 self._consecutive = 0
254#: Isolate provider calls from the user's Claude Code configuration: no MCP
255#: servers (--strict-mcp-config with none given) and no hooks.
256HERMETIC_ARGS: tuple[str, ...] = ("--strict-mcp-config", "--settings", '{"disableAllHooks": true}')
258#: Disable Claude Code's built-in tools. Separate from HERMETIC_ARGS on
259#: purpose: that public constant shipped in 0.16.0 and means "isolate from the
260#: USER's Claude Code config" (MCP servers, hooks); changing its value is a
261#: breaking change under the additive-first API policy (griffe). Built-in tools
262#: are Claude Code's own defaults, a different concern. This provider never
263#: uses them (``complete(tools=...)`` raises NotImplementedError), yet their
264#: definitions were ~26,500 of the ~33,000 input tokens on every call
265#: (measured 2026-10-03: 33,142 -> 6,676 with the flag), and a model that
266#: tried one hit ``--max-turns 1`` and failed with "Reached max turns (1)"
267#: (observed in a benchmark run).
268_BUILTIN_TOOLS_OFF: tuple[str, ...] = ("--tools", "")
271class ClaudeCliProvider:
272 """LLMProvider whose ``complete()`` shells out to the Claude Code CLI."""
274 SPI_VERSION: ClassVar[int] = 1
276 def __init__(
277 self,
278 *,
279 model: str = "haiku",
280 binary: str | None = None,
281 timeout: float = 180.0,
282 max_retries: int = 4,
283 max_concurrency: int | None = None,
284 rate_limit_backoff: float | None = None,
285 max_thinking_tokens: int | None = None,
286 ) -> None:
287 resolved = binary or os.environ.get("CLAUDE_CLI_BIN") or shutil.which("claude")
288 if not resolved:
289 raise RuntimeError(
290 "ClaudeCliProvider requires the Claude Code CLI on PATH "
291 "(or CLAUDE_CLI_BIN=/path/to/claude)."
292 )
293 self._bin = resolved
294 self._model = model
295 self._timeout = timeout
296 self._max_retries = max(1, max_retries)
297 # Throttling clears on a minutes-scale window; seconds-scale retries
298 # just burn the budget inside it. First delay, doubled per attempt.
299 self._rate_limit_backoff = rate_limit_backoff or float(
300 os.environ.get("ASTROCYTE_CLAUDE_CLI_RATE_LIMIT_BACKOFF", "60")
301 )
302 conc = max_concurrency or int(
303 os.environ.get("ASTROCYTE_CLAUDE_CLI_MAX_CONCURRENCY", "4")
304 )
305 self._sem = asyncio.Semaphore(max(1, conc))
306 # Extended-thinking budget passed to the CLI as MAX_THINKING_TOKENS.
307 # ``None`` leaves Claude Code's default (thinking ON); ``0`` turns it
308 # off. Measured on structured extraction (18 real LongMemEval batches,
309 # 2026-10-03): ~85% of output tokens were hidden thinking; with it off
310 # a call took 31 s instead of 68 s with 55% fewer output tokens, and —
311 # once metadata was joined by chunk_index — no alignment or entity-
312 # grounding penalty. Default unchanged because that evidence covers
313 # extraction, not reflect/synthesis quality.
314 if max_thinking_tokens is None:
315 env_budget = os.environ.get("ASTROCYTE_CLAUDE_CLI_MAX_THINKING_TOKENS")
316 max_thinking_tokens = int(env_budget) if env_budget not in (None, "") else None
317 if max_thinking_tokens is not None and max_thinking_tokens < 0:
318 raise ValueError(f"max_thinking_tokens must be >= 0, got {max_thinking_tokens}")
319 self._max_thinking_tokens = max_thinking_tokens
320 # Private empty cwd: print-mode runs must not pick up any project's
321 # CLAUDE.md context.
322 self._cwd = tempfile.mkdtemp(prefix="astrocyte-claude-cli-")
323 self._warned_model_override = False
324 # Shared across every call on this provider: a failure burst pauses
325 # the whole pipeline instead of letting it grind through with broken
326 # ingest. Tunable for tests via the env vars below.
327 self._breaker = _CircuitBreaker(
328 threshold=int(os.environ.get("ASTROCYTE_CLAUDE_CLI_BREAKER_THRESHOLD", "5")),
329 cooldown=float(os.environ.get("ASTROCYTE_CLAUDE_CLI_BREAKER_COOLDOWN", "300")),
330 )
332 def capabilities(self) -> LLMCapabilities:
333 return LLMCapabilities(
334 supports_multimodal_completion=False,
335 modalities_supported=("text",),
336 supports_multimodal_embedding=False,
337 supports_batch_embed=False,
338 )
340 async def complete(
341 self,
342 messages: list[Message],
343 model: str | None = None,
344 max_tokens: int = 1024, # noqa: ARG002 — CLI limitation, see module docstring
345 temperature: float = 0.0, # noqa: ARG002 — CLI limitation
346 tools: list[ToolDefinition] | None = None,
347 tool_choice: str | None = None, # noqa: ARG002
348 response_format: dict | None = None,
349 ) -> Completion:
350 if tools:
351 raise NotImplementedError(
352 "ClaudeCliProvider does not support native function calling "
353 "(tools=...). Use an API-backed provider for agentic paths."
354 )
355 # Per-call model overrides carry OpenAI names from legacy call sites
356 # (e.g. summarizer_model='gpt-4o-mini'); the CLI model wins. Warn once.
357 if model and model != self._model and not self._warned_model_override:
358 logger.warning(
359 "claude_cli: ignoring per-call model override %r — using %r "
360 "(further overrides silenced)", model, self._model,
361 )
362 self._warned_model_override = True
364 prompt = _render_prompt(messages, response_format)
365 env = {k: v for k, v in os.environ.items() if k != "ANTHROPIC_API_KEY"}
366 # This is an LLM used as an API: it must not run the user's hooks or
367 # load their MCP servers. Without this, every call loaded whatever MCP
368 # servers the user had registered (changing the prompt) and ran their
369 # hooks — including Astrocyte's own capture hooks, which then captured
370 # the daemon's internal prompts and triggered more calls (observed).
371 env["ASTROCYTE_HOOKS"] = "off"
372 if self._max_thinking_tokens is not None:
373 env["MAX_THINKING_TOKENS"] = str(self._max_thinking_tokens)
375 last_error: str = ""
376 rate_limited = False
377 attempt = 0
378 while attempt < self._max_retries:
379 # A sustained failure burst pauses every caller here rather than
380 # letting the pipeline grind on with failing ingest.
381 await self._breaker.await_closed()
382 try:
383 # The prompt goes in through a file, not a pipe written after
384 # spawn: `claude -p` waits only 3 s for stdin and then fails
385 # with "no stdin data received". Any event-loop stall between
386 # spawn and the write (a CPU-bound embedding, a GC pause) lost
387 # that race about once per benchmark chunk, and each failure
388 # fed the rate-limit breaker. A file has the whole prompt
389 # readable at exec. TemporaryFile is unlinked on POSIX, so the
390 # prompt (user memory) never sits on disk under a name.
391 async with self._sem:
392 with tempfile.TemporaryFile() as prompt_file:
393 prompt_file.write(prompt.encode("utf-8"))
394 prompt_file.flush()
395 prompt_file.seek(0)
396 proc = await asyncio.create_subprocess_exec(
397 self._bin,
398 "-p",
399 "--model", self._model,
400 "--output-format", "text",
401 "--max-turns", "1",
402 *HERMETIC_ARGS,
403 *_BUILTIN_TOOLS_OFF,
404 stdin=prompt_file,
405 stdout=asyncio.subprocess.PIPE,
406 stderr=asyncio.subprocess.PIPE,
407 cwd=self._cwd,
408 env=env,
409 )
410 try:
411 stdout, stderr = await asyncio.wait_for(
412 proc.communicate(),
413 timeout=self._timeout,
414 )
415 except asyncio.TimeoutError:
416 proc.kill()
417 await proc.wait()
418 raise
419 if proc.returncode == 0:
420 text = stdout.decode("utf-8", "replace").strip()
421 if response_format is not None:
422 # Emulated JSON mode: normalize fences/prose to bare
423 # JSON so downstream ``json.loads`` behaves as it
424 # would under native constrained decoding. An
425 # unparseable response is retried; after the final
426 # attempt we fall back to the RAW text (never raise)
427 # — pipeline stages have tolerant parsing and their
428 # own graceful-degradation paths.
429 normalized = _normalize_json_output(text)
430 if normalized is not None:
431 text = normalized
432 elif attempt < self._max_retries - 1:
433 last_error = "unparseable JSON-mode output"
434 logger.warning(
435 "claude_cli: JSON-mode output unparseable "
436 "(attempt %d/%d) — retrying",
437 attempt + 1, self._max_retries,
438 )
439 attempt += 1
440 await asyncio.sleep(2 * attempt)
441 continue
442 else:
443 logger.warning(
444 "claude_cli: JSON-mode output unparseable "
445 "after %d attempts — returning raw text",
446 self._max_retries,
447 )
448 await self._breaker.record_success()
449 return Completion(
450 text=text,
451 model=self._model,
452 usage=None, # CLI print mode reports no token usage
453 tool_calls=None,
454 )
455 # Non-zero exit. The CLI reports throttling on STDOUT, so both
456 # streams are captured — logging only stderr is what made the
457 # first claude-native run's 1329 failures look empty.
458 out = stdout.decode("utf-8", "replace").strip()
459 err = stderr.decode("utf-8", "replace").strip()
460 combined = (err + "\n" + out).strip()
461 rate_limited = _looks_rate_limited(combined)
462 last_error = (combined or f"exit {proc.returncode}, no output")[:500]
463 logger.warning(
464 "claude_cli complete attempt %d/%d exited %s%s: %s",
465 attempt + 1, self._max_retries, proc.returncode,
466 " [RATE-LIMITED]" if rate_limited else "", last_error,
467 )
468 except asyncio.TimeoutError:
469 last_error = f"timeout after {self._timeout}s"
470 rate_limited = False
471 logger.warning(
472 "claude_cli complete attempt %d/%d timed out",
473 attempt + 1, self._max_retries,
474 )
475 except Exception as exc: # noqa: BLE001
476 last_error = str(exc)[:500]
477 rate_limited = _looks_rate_limited(last_error)
478 logger.warning(
479 "claude_cli complete attempt %d/%d failed: %s",
480 attempt + 1, self._max_retries, exc,
481 )
483 attempt += 1
484 if attempt < self._max_retries:
485 if rate_limited:
486 # Throttling clears in minutes, not seconds. The old
487 # 2s/4s ladder burned all retries inside a window that
488 # had barely started.
489 delay = min(self._rate_limit_backoff * (2 ** (attempt - 1)), 600.0)
490 logger.warning(
491 "claude_cli: rate limited — backing off %.0fs before "
492 "attempt %d/%d", delay, attempt + 1, self._max_retries,
493 )
494 else:
495 delay = 2.0 * attempt
496 await asyncio.sleep(delay)
498 await self._breaker.record_failure()
499 raise RuntimeError(
500 f"claude_cli complete failed after {self._max_retries} attempts"
501 f"{' (rate limited)' if rate_limited else ''}: {last_error}"
502 )
504 async def embed(
505 self,
506 texts: list[str],
507 model: str | None = None,
508 ) -> list[list[float]]:
509 raise NotImplementedError(
510 "The Claude Code CLI has no embeddings surface. Compose "
511 "ClaudeCliProvider with LocalEmbeddingsProvider via "
512 "CompositeLLMProvider (astrocyte.providers.composite)."
513 )