fix(claude_desktop): guard env parsing and truncate file-list arrays in headroom hook

Two review findings on PR #2871:

- coderabbitai: MIN_CHARS/MIN_SAVED_TOKENS parsed with a bare int() at module
  import time, before any try/except could catch a bad value — a malformed
  env_vars passthrough would crash the hook on every matched tool call instead
  of failing open as documented. Wrapped in _int_env() with a safe fallback.

- chatgpt-codex-connector: Glob and Grep (files_with_matches mode) return a
  `filenames: string[]` field per the CLI's own output schema, which the
  hook's string-only candidate scan never touched — large file listings, the
  exact case named in the CLAUDE.md guidance this add-on installs, passed
  through uncompressed. Verified empirically that routing such an array
  through compress()/SmartCrusher (as done for JSON-blob string fields)
  silently subsamples it — 600 paths collapsed to ~15 with no visible marker,
  unsafe for paths the model needs to act on individually. Added a separate
  deterministic path: arrays over ARRAY_KEEP (40) entries are truncated in
  order with one labeled marker entry appended, full array recoverable from
  the CCR store by hash. Verified round-trip on Glob- and Grep-shaped
  payloads (600 and 200 entries); confirmed order preservation and that
  small arrays still pass through untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
alexbelgium
2026-07-16 14:46:10 +02:00
parent 9bbd72e70a
commit 26101a1104

View File

@@ -24,14 +24,31 @@ Design constraints:
the savings on tool output anyway; plain prose passes through unchanged.
- stderr fields are never compressed — error text must reach the model verbatim
(matching Headroom's own error-protection policy).
- File-list arrays (Glob's `filenames`, Grep's `filenames` in files_with_matches
mode — both typed `string[]` by the CLI's own output schema) are handled
separately from prose/JSON-blob fields: Headroom's SmartCrusher subsamples
JSON arrays for informational dumps, which is fine for e.g. a list of sensor
states but silently drops most paths from a file listing the model needs to
act on. Those fields are truncated deterministically instead (keep the first
N entries, append one marker string) so the model always sees a labeled cut
point rather than a shorter list it might mistake for the complete result.
"""
import json
import os
import sys
MIN_CHARS = int(os.environ.get("HEADROOM_HOOK_MIN_CHARS", "4000"))
MIN_SAVED_TOKENS = int(os.environ.get("HEADROOM_HOOK_MIN_SAVED_TOKENS", "50"))
def _int_env(name: str, default: str) -> int:
try:
return int(os.environ.get(name, default))
except (TypeError, ValueError):
return int(default)
MIN_CHARS = _int_env("HEADROOM_HOOK_MIN_CHARS", "4000")
MIN_SAVED_TOKENS = _int_env("HEADROOM_HOOK_MIN_SAVED_TOKENS", "50")
ARRAY_KEEP = _int_env("HEADROOM_HOOK_ARRAY_KEEP", "40")
TTL_SECONDS = 3600 # matches the headroom MCP server's session TTL
SKIP_KEYS = {"stderr"}
@@ -57,18 +74,26 @@ def main() -> int:
return 0
response = payload.get("tool_response")
# Find big string fields before paying the headroom import cost.
# Find big string/array fields before paying the headroom import cost.
def is_string_array(value):
return isinstance(value, list) and len(value) > ARRAY_KEEP and all(isinstance(v, str) for v in value)
if isinstance(response, str):
candidates = ["__whole__"] if len(response) >= MIN_CHARS else []
string_candidates = ["__whole__"] if len(response) >= MIN_CHARS else []
array_candidates = []
elif isinstance(response, dict):
candidates = [
string_candidates = [
key
for key, value in response.items()
if key not in SKIP_KEYS and isinstance(value, str) and len(value) >= MIN_CHARS
]
array_candidates = [
key for key, value in response.items() if key not in SKIP_KEYS and is_string_array(value)
]
else:
candidates = []
if not candidates:
string_candidates = []
array_candidates = []
if not string_candidates and not array_candidates:
return 0
# Keep Kompress's cache probe away from the tmpfs-backed ~/.cache default.
@@ -111,17 +136,56 @@ def main() -> int:
f"call mcp__headroom__headroom_retrieve with hash={hash_key} if you need the full original]"
)
def shrink_array(items):
nonlocal store
original_json = json.dumps(items)
if len(original_json) < MIN_CHARS:
return None
kept = items[:ARRAY_KEEP]
truncated_json = json.dumps(kept)
# No ML/token-counter call needed for a plain truncation decision; a char/4
# estimate is the same fallback Headroom's own cost estimator uses and is
# only used here to decide eligibility and annotate the marker.
tokens_before = max(1, len(original_json) // 4)
tokens_after = max(1, len(truncated_json) // 4)
if tokens_before - tokens_after < MIN_SAVED_TOKENS:
return None
if store is None:
store = get_compression_store()
hash_key = store.store(
original=original_json,
compressed=truncated_json,
original_tokens=tokens_before,
compressed_tokens=tokens_after,
compression_strategy="posttooluse_hook_array_truncate",
ttl=TTL_SECONDS,
)
totals[0] += tokens_before
totals[1] += tokens_after
remaining = len(items) - len(kept)
marker = (
f"[headroom: {remaining} more of {len(items)} entries omitted "
f"({tokens_before}->{tokens_after} tokens); call mcp__headroom__headroom_retrieve "
f"with hash={hash_key} for the complete list]"
)
return kept + [marker]
updated = None
if isinstance(response, str):
updated = shrink(response)
else:
rewritten = dict(response)
changed = False
for key in candidates:
for key in string_candidates:
new_value = shrink(rewritten[key])
if new_value is not None:
rewritten[key] = new_value
changed = True
for key in array_candidates:
new_value = shrink_array(rewritten[key])
if new_value is not None:
rewritten[key] = new_value
changed = True
if changed:
updated = rewritten
if updated is None: