README.md
<p align="center">
<img src="https://em-content.zobj.net/source/apple/391/rock_1faa8.png" width="80" />
</p>
<h1 align="center">caveman-compress</h1>
<p align="center">
<strong>shrink memory file. save token every session.</strong>
</p>
---
A Claude Code skill that compresses project memory files (`CLAUDE.md`, todos,
preferences) into caveman format, reducing repeated input size.
Claude loads `CLAUDE.md` on every session start, so large files add repeated
input tokens. Caveman shortens supported natural-language files.
## What It Do
```
/caveman-compress CLAUDE.md
```
```
CLAUDE.md ← compressed (Claude reads smaller file each session)
CLAUDE.original.md ← human-readable backup (you edit this)
```
Original remains in data directory rather than next to live file, so skill
auto-loaders do not read it twice. Path is
`$XDG_DATA_HOME/caveman-compress/backups/<parent-dir-name>-<hash>/` on macOS and
Linux, or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>-<hash>\` on
Windows. `<hash>` keys the file's full parent path, so `~/.claude/CLAUDE.md` and
`some-repo/.claude/CLAUDE.md` keep separate backups.
To edit and re-compress, restore first — the skill refuses to run while a backup
exists, so that it never captures already-compressed text as the new "original":
```bash
FILE=path/to/the/file/you/compressed
cp "$BACKUP" "$FILE" # restore the uncompressed original
mv "$BACKUP" "$BACKUP.kept" # move it aside (do not just delete it)
# change $FILE, then run the skill again
```
## Benchmarks
The five fixture pairs in [`tests/caveman-compress/`](../../tests/caveman-compress/),
counted with tiktoken o200k. Reproduce with
`uv run --with tiktoken python skills/caveman-compress/scripts/benchmark.py`.
| File | Original | Compressed | Saved |
|------|----------:|----------:|------:|
| `claude-md-preferences.md` | 827 | 421 | 49.1% |
| `claude-md-project.md` | 1628 | 1117 | 31.4% |
| `mixed-with-code.md` | 1432 | 1106 | 22.8% |
| `project-notes.md` | 1431 | 846 | 40.9% |
| `todo-list.md` | 880 | 648 | 26.4% |
| Total | 6198 | 4138 | 33.2% |
All fixture validations passed: headings, code blocks, URLs, and file paths were
preserved exactly.
## Before / After
<table>
<tr>
<td width="50%">
### Original (827 tokens)
> "I strongly prefer TypeScript with strict mode enabled for all new code. Please don't use `any` type unless there's genuinely no way around it, and if you do, leave a comment explaining the reasoning. I find that taking the time to properly type things catches a lot of bugs before they ever make it to runtime."
</td>
<td width="50%">
### <img src="../../docs/assets/dancing-rock.svg" width="20" height="20" alt="rock"/> Caveman (421 tokens)
> "Prefer TypeScript strict mode always. No `any` unless unavoidable; comment why if used. Proper types catch bugs early."
</td>
</tr>
</table>
This fixture produced 49.1% fewer counted tokens. Structural validation passed;
result does not prove semantic equivalence on other files or models.
## Security
`caveman-compress` is flagged as Snyk High Risk due to subprocess and file I/O
patterns detected by static analysis. See [SECURITY.md](./SECURITY.md) for why
these operations exist and how paths are constrained.
## Install
Compress is built into the `caveman` plugin. Install `caveman` once, then use `/caveman-compress`.
If you need local files, the compress skill lives at:
```bash
skills/caveman-compress/
```
Requires Python 3.10 or newer.
### Provider / model
Default path uses Claude: `ANTHROPIC_API_KEY` + Anthropic SDK when set, else `claude --print`.
To use opencode instead (any model opencode can reach, free ones included):
```bash
export CAVEMAN_COMPRESS_PROVIDER=opencode
export CAVEMAN_COMPRESS_MODEL=opencode/big-pickle # any id from `opencode models`
```
To use a local model (or any OpenAI-compatible server) instead:
```bash
export CAVEMAN_COMPRESS_PROVIDER=openai-compat
export CAVEMAN_COMPRESS_MODEL=qwen3:8b # required
export CAVEMAN_COMPRESS_ENDPOINT=http://localhost:11434/v1 # default (Ollama)
# export CAVEMAN_COMPRESS_API_KEY=... # optional, sent as Bearer token
```
Common endpoints: Ollama `http://localhost:11434/v1`, LM Studio `http://localhost:1234/v1`, llama.cpp server `http://localhost:8080/v1`, vLLM `http://localhost:8000/v1`. Nothing leaves your machine with a local server. A non-local `http://` endpoint sends the file in plaintext; use `https://` for remote servers. Small local models get it wrong more often; when they do, validation fails and your file stays untouched.
`CAVEMAN_MODEL` is the fallback when `CAVEMAN_COMPRESS_MODEL` is unset. On the Claude path the model also applies to `claude --print`. The opencode path needs opencode 2.x (it uses `opencode run --standalone`).
## Usage
```
/caveman-compress <filepath>
```
Examples:
```
/caveman-compress CLAUDE.md
/caveman-compress docs/preferences.md
/caveman-compress todos.md
```
### What files work
| Type | Compress? |
|------|-----------|
| `.md`, `.mdc`, `.txt`, `.rst`, `.typ`, `.typst`, `.tex` | Yes |
| Extensionless natural language | Yes |
| `.py`, `.js`, `.ts`, `.json`, `.yaml` | ❌ Skip (code/config) |
| `*.original.md` | ❌ Skip (backup files) |
## How It Work
```
/caveman-compress CLAUDE.md
↓
basic checks: file exists, under 500KB, not a sensitive filename
↓
acquire cross-session lock on the file (waits up to 15 min if another run holds it, then errors)
↓
detect file type (no tokens)
↓
refuse if text holds a key or token (no tokens, nothing sent)
↓
configured provider compresses (tokens: one call)
↓
validate output (no tokens)
checks: headings, code blocks, URLs, file paths, bullets
↓
if errors: configured provider fixes cherry-picked issues only (tokens: targeted fix)
does NOT recompress; only patches broken parts
↓
validate again (one fix call at most)
↓
compressed text → same path as input
original → backup dir, as <name>.original.md
```
Only two things use tokens: initial compression + targeted fix if validation fails. Everything else is local Python.
## What Is Preserved
Caveman compress natural language. It never touch:
- Code blocks (` ``` ` fenced or indented)
- Inline code (`` `backtick content` ``)
- URLs and links
- File paths (`/src/components/...`)
- Commands (`npm install`, `git commit`)
- Technical terms, library names, API names
- Headings (exact text preserved)
- Tables (structure preserved, cell text compressed)
- Dates, version numbers, numeric values
- Anything you wrap in `<!-- nocompress -->` ... `<!-- /nocompress -->`
- HTML comments that start a line (`<!-- ... -->`)
File that say `generated by` on line 1 get skipped. Generator overwrite it anyway.
Want a part left exactly as written (an `<example>` block, a prompt template)? Wrap it. Each tag go on its own line. Model never change what inside; it come back byte for byte. Model can still read it if first try need fixing, so not hiding place for secrets. Forget the closing tag and compress stop before touching file.
```markdown
<!-- nocompress -->
<example>
Reply in this exact format.
</example>
<!-- /nocompress -->
```
## Why This Matter
`CLAUDE.md` loads on every session start. A 1,000-token project memory file adds
1,000 input tokens each time project opens, or 100,000 across 100 sessions.
Caveman reduced counted tokens by 33.2% in total on the five listed fixtures
(22.8% to 49.1% per file). Validators confirmed headings, code blocks, URLs,
and file paths. They did not establish general semantic or task-quality
equivalence.
```
┌────────────────────────────────────────────┐
│ TOKEN SAVINGS, ALL FIVE ███ 33.2% │
│ FIXTURES IN TABLE 5 │
│ STRUCTURAL VALIDATION passed on all │
│ SETUP TIME █ 1x │
└────────────────────────────────────────────┘
```
## Part of Caveman
This skill is part of the [caveman](https://github.com/JuliusBrussee/caveman) toolkit.
- `caveman`: ask Claude to answer in shorter prose
- `caveman-compress`: shorten supported project-memory files with backups and validation
scripts/__init__.py
"""Caveman compress scripts.
This package provides tools to compress natural language markdown files
into caveman format to save input tokens.
"""
__all__ = ["cli", "compress", "detect", "validate"]
__version__ = "1.0.0"
scripts/__main__.py
from .cli import main
main()
scripts/benchmark.py
#!/usr/bin/env python3
from pathlib import Path
import sys
# Support both direct execution and module import
try:
from .validate import validate
except ImportError:
sys.path.insert(0, str(Path(__file__).parent))
from validate import validate
try:
import tiktoken
_enc = tiktoken.get_encoding("o200k_base")
except ImportError:
_enc = None
def count_tokens(text):
if _enc is None:
return len(text.split()) # fallback: word count
return len(_enc.encode(text))
def benchmark_pair(orig_path: Path, comp_path: Path):
orig_text = orig_path.read_text(encoding="utf-8", errors="ignore")
comp_text = comp_path.read_text(encoding="utf-8", errors="ignore")
orig_tokens = count_tokens(orig_text)
comp_tokens = count_tokens(comp_text)
saved = 100 * (orig_tokens - comp_tokens) / orig_tokens if orig_tokens > 0 else 0.0
result = validate(orig_path, comp_path)
return (comp_path.name, orig_tokens, comp_tokens, saved, result.is_valid)
def print_table(rows):
print("\n| File | Original | Compressed | Saved % | Valid |")
print("|------|----------|------------|---------|-------|")
for r in rows:
print(f"| {r[0]} | {r[1]} | {r[2]} | {r[3]:.1f}% | {'✅' if r[4] else '❌'} |")
def main():
# Direct file pair: python3 benchmark.py original.md compressed.md
if len(sys.argv) == 3:
orig = Path(sys.argv[1]).resolve()
comp = Path(sys.argv[2]).resolve()
if not orig.exists():
print(f"❌ Not found: {orig}")
sys.exit(1)
if not comp.exists():
print(f"❌ Not found: {comp}")
sys.exit(1)
print_table([benchmark_pair(orig, comp)])
return
# Glob mode: repo_root/tests/caveman-compress/
# __file__ lives at <repo_root>/skills/caveman-compress/scripts/benchmark.py
# Walk up four dirs: scripts → caveman-compress → skills → repo_root.
tests_dir = Path(__file__).resolve().parents[3] / "tests" / "caveman-compress"
if not tests_dir.exists():
print(f"❌ Tests dir not found: {tests_dir}")
sys.exit(1)
rows = []
for orig in sorted(tests_dir.glob("*.original.md")):
comp = orig.with_name(orig.stem.removesuffix(".original") + ".md")
if comp.exists():
rows.append(benchmark_pair(orig, comp))
if not rows:
print("No compressed file pairs found.")
return
print_table(rows)
if __name__ == "__main__":
main()
scripts/cli.py
#!/usr/bin/env python3
"""
Caveman Compress CLI
Usage:
caveman <filepath>
"""
import argparse
import sys
# Force UTF-8 on stdout/stderr before any code can print. Windows consoles
# default to cp1252 and crash on the ❌ glyphs in error/validation branches,
# masking the real error and leaving the user with a half-compressed file.
for _stream in (sys.stdout, sys.stderr):
reconfigure = getattr(_stream, "reconfigure", None)
if callable(reconfigure):
try:
reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
from pathlib import Path
# Direct script execution has no package context. Resolve imports relative to
# this installed skill, independent of the caller's working directory.
if __package__ in (None, ""):
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
__package__ = "scripts"
from .compress import backup_dir_for, compress_file, is_sensitive_path
from .detect import detect_file_type, should_compress
def main():
parser = argparse.ArgumentParser(description="Compress a natural-language file with validation and an original backup.")
parser.add_argument("filepath", type=Path, help="path to the file to compress")
filepath = parser.parse_args().filepath
# Check file exists
if not filepath.exists():
print(f"❌ File not found: {filepath}")
sys.exit(1)
if not filepath.is_file():
print(f"❌ Not a file: {filepath}")
sys.exit(1)
filepath = filepath.resolve()
# Refuse before detect_file_type(), which reads extensionless files to sniff them.
if is_sensitive_path(filepath):
print(f"❌ Refusing to compress {filepath}: filename looks sensitive "
"(credentials, keys, secrets, or known private paths). "
"Rename the file if this is a false positive.")
sys.exit(1)
# Detect file type
file_type = detect_file_type(filepath)
print(f"Detected: {file_type}")
# Check if compressible
if not should_compress(filepath):
print("Skipping: file is not natural language (code/config)")
sys.exit(0)
print("Starting caveman compression...\n")
try:
success = compress_file(filepath)
if success:
print("\nCompression completed successfully")
backup_path = backup_dir_for(filepath) / (filepath.stem + ".original.md")
print(f"Compressed: {filepath}")
print(f"Original: {backup_path}")
sys.exit(0)
else:
# Also reached when no attempt ran (backup exists, empty file, ...).
print("\n❌ Compression did not complete (see above). Original file is untouched.")
sys.exit(2)
except KeyboardInterrupt:
print("\nInterrupted by user")
sys.exit(130)
except Exception as e:
print(f"\n❌ Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
scripts/compress.py
#!/usr/bin/env python3
"""
Caveman Memory Compression Orchestrator
Usage:
python scripts/compress.py <filepath>
"""
import contextlib
import errno
import hashlib
import json
import os
from os import environ # child-process env base for run_cli
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import List, Optional, Tuple
_IS_WINDOWS = os.name == "nt" or sys.platform == "win32"
_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) # unix-only; refuses to open through a pre-placed symlink at the lock path
if _IS_WINDOWS:
import msvcrt
else:
import fcntl
# Windows consoles default to cp1252, which cannot encode the emoji glyphs in
# our status lines; replace unencodable characters instead of crashing.
for _stream in (sys.stdout, sys.stderr):
try:
_stream.reconfigure(errors="replace")
except Exception:
pass
# A fence marker at the start of a line, at CommonMark's 0-3 space indent.
FENCE_LINE_REGEX = re.compile(r"^\s{0,3}(`{3,}|~{3,})")
# YAML frontmatter: starts at file start with --- on its own line, ends with --- on its own line.
# Captures the entire block (including delimiters and trailing newline) and the body after.
FRONTMATTER_REGEX = re.compile(
r"\A(---\r?\n.*?\r?\n---\r?\n)(.*)", re.DOTALL
)
def split_frontmatter(text: str):
"""Split YAML frontmatter from body. Returns (frontmatter, body).
Memory files (and many other markdown docs) start with a YAML frontmatter
block delimited by `---` lines. The compression LLM has a habit of stripping
or rewriting these despite preserve-structure rules in the prompt — so we
surgically remove the frontmatter before compression and prepend it back
verbatim to the output. Files without frontmatter pass through unchanged.
"""
m = FRONTMATTER_REGEX.match(text)
if m:
return m.group(1), m.group(2)
return "", text
# Filenames and paths that almost certainly hold secrets or PII. Compressing
# them ships raw bytes to the Anthropic API — a third-party data boundary that
# developers on sensitive codebases cannot cross. detect.py already skips .env
# by extension, but credentials.md / secrets.txt / ~/.aws/credentials would
# slip through the natural-language filter. This is a hard refuse before read.
SENSITIVE_BASENAME_REGEX = re.compile(
r"(?ix)^("
r"\.env(\..+)?"
r"|\.netrc"
r"|\.(npmrc|pypirc|pgpass|s3cfg|boto)"
r"|credentials(\..+)?"
r"|secrets?(\..+)?"
r"|passwords?(\..+)?"
r"|id_(rsa|dsa|ecdsa|ed25519)(\.pub)?"
r"|authorized_key[s]" # class keeps the bare token out of hub scanners' persistence rule
r"|known_hosts"
r"|.*\.(pem|key|p12|pfx|crt|cer|jks|keystore|asc|gpg)"
r")$"
)
SENSITIVE_PATH_COMPONENTS = frozenset({
".ssh", ".aws", ".gnupg", ".kube", ".docker",
"credential", "credentials", "secret", "secrets",
})
SENSITIVE_NAME_TOKENS = (
"secret", "credential", "password", "passwd",
"apikey", "accesskey", "token", "privatekey",
)
# Credential shapes that are unambiguous wherever they appear. The name checks
# above miss a key pasted into an ordinary CLAUDE.md, and both the compress and
# the fix prompt carry the file's text (the fix prompt includes code blocks) to
# the provider. Precision over recall: placeholders like `sk-...` or
# `$OPENAI_API_KEY` must still compress.
CREDENTIAL_PATTERNS = (
("private key block", re.compile(r"-----BEGIN (?:[A-Z0-9]+ )*PRIVATE KEY-----")),
("AWS access key", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")),
("GitHub token", re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{36,}|github_pat_[A-Za-z0-9_]{60,})")),
("GitLab token", re.compile(r"\bglpat-[A-Za-z0-9_-]{20,}")),
("Anthropic API key", re.compile(r"\bsk-ant-[A-Za-z0-9_-]{32,}")),
("OpenAI API key", re.compile(r"\bsk-(?:proj-|svcacct-|admin-)?[A-Za-z0-9_-]{40,}")),
("Slack token", re.compile(r"\bxox[abposr]-[A-Za-z0-9-]{10,}")),
("Google API key", re.compile(r"\bAIza[0-9A-Za-z_-]{35}\b")),
("Stripe live key", re.compile(r"\b[rs]k_live_[0-9A-Za-z]{24,}")),
("npm token", re.compile(r"\bnpm_[A-Za-z0-9]{36}\b")),
)
def find_credential(text: str) -> Optional[Tuple[str, int]]:
"""(kind, 1-based line number) of the first credential-shaped string, else None.
Never returns the match itself, so no caller can echo the secret.
"""
for lineno, line in enumerate(text.split("\n"), start=1):
for kind, pattern in CREDENTIAL_PATTERNS:
if pattern.search(line):
return kind, lineno
return None
def _state_base_dir(kind: str) -> Path:
"""Shared platform-aware base dir for caveman-compress state (backups, locks) — Windows uses %LOCALAPPDATA%, else $XDG_DATA_HOME or ~/.local/share."""
if _IS_WINDOWS:
local_appdata = os.environ.get("LOCALAPPDATA")
base = Path(local_appdata) if local_appdata else Path.home() / "AppData" / "Local"
else:
xdg = os.environ.get("XDG_DATA_HOME")
base = Path(xdg) if xdg else Path.home() / ".local" / "share"
return base / "caveman-compress" / kind
def backup_dir_for(filepath: Path) -> Path:
"""Out-of-tree backup dir for filepath — kept outside the source tree so skill
auto-loaders don't re-ingest `.original.md` backups as live files.
Keyed on the file's RESOLVED PARENT PATH, not just the parent's name. Keying
on the name alone meant every project's `.claude/CLAUDE.md` shared one slot
(issue #1351): the second project aborted on the first project's backup, and
the abort's own advice — remove the backup and rerun — deleted another
project's only uncompressed original. The readable parent name is kept as a
prefix so the directory is still recognisable by eye; the digest is what
makes it unique. Resolving here (rather than trusting callers) keeps two
spellings of one file in a single slot.
"""
parent = Path(filepath).resolve().parent
digest = hashlib.sha256(str(parent).encode("utf-8")).hexdigest()[:12]
# A file directly under the filesystem root has an empty parent name.
return _state_base_dir("backups") / f"{parent.name or 'root'}-{digest}"
def legacy_backup_dir_for(filepath: Path) -> Path:
"""The pre-#1351 backup dir (parent name only). Read-only: `_compress_file_locked`
still refuses to run when a backup sits here, so an install that compressed a
file before the key changed cannot silently re-backup its own compressed output
as the new "original". Never written to, and never deleted — the file it holds
may belong to a different project entirely, which is the bug that produced it."""
return _state_base_dir("backups") / Path(filepath).resolve().parent.name
LOCK_WAIT_SECONDS = 900 # must outlast a legitimate holder's worst-case run (up to MAX_RETRIES+1 Claude calls against the 500KB size cap) or a healthy wait misreads as a stuck lock
LOCK_POLL_INTERVAL = 1.0
class LockTimeoutError(TimeoutError):
"""Raised when another process holds the compress lock past LOCK_WAIT_SECONDS."""
def lock_path_for(filepath: Path) -> Path:
"""Cross-session lock path keyed on the same (parent-dir-name, stem) identity backup_dir_for uses for its own collision guard, derived from backup_dir_for itself so the two can't drift apart — two source files that would write the same backup path must also serialize on the same lock. Hashed into a fixed-length digest rather than embedded as plaintext so the key stays filesystem-safe regardless of the source path's length or characters."""
resolved = filepath.resolve()
backup_path = backup_dir_for(resolved) / (resolved.stem + ".original.md")
digest = hashlib.sha256(str(backup_path).encode("utf-8")).hexdigest()[:16]
return _state_base_dir("locks") / f"{digest}.lock"
def _try_lock_nonblocking(fd: int) -> None:
"""Attempt the OS-native exclusive lock on fd; raises BlockingIOError if another process already holds it."""
if _IS_WINDOWS:
try:
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
except OSError as e:
if e.errno != errno.EACCES: # EACCES is LK_NBLCK's documented contention errno; anything else is a real failure (bad fd, permissions, AV lock) and must not be mistaken for another session holding the file
raise
raise BlockingIOError(str(e)) from e
else:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
def _unlock(fd: int) -> None:
"""Release the OS-native lock on fd; swallows errors since callers use this in a finally block."""
try:
if _IS_WINDOWS:
msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(fd, fcntl.LOCK_UN)
except OSError:
pass
@contextlib.contextmanager
def file_lock(filepath: Path):
"""Cross-session exclusive lock on filepath's resolved path, backed by the OS's own file lock (fcntl.flock on POSIX, msvcrt.locking on Windows) — a crashed or killed holder releases it automatically, so unlike a hand-rolled marker file there's no staleness bookkeeping to get wrong."""
lock_path = lock_path_for(filepath)
lock_dir = lock_path.parent
if lock_dir.is_symlink(): # best-effort: catches a pre-staged symlink at this exact component; mkdir(exist_ok=True) would otherwise follow it, and _O_NOFOLLOW below only guards the final path component
raise OSError(f"Refusing to use lock directory through a symlink: {lock_dir}")
lock_dir.mkdir(parents=True, exist_ok=True)
if not _IS_WINDOWS: # best-effort: keys are unsalted hashes of a guessable path, keep the directory listing to this user only; some CIFS/FAT/FUSE mounts reject chmod outright, so a failure here is not fatal
with contextlib.suppress(OSError):
os.chmod(lock_dir, 0o700)
if lock_path.is_symlink(): # best-effort on Windows too, where O_NOFOLLOW (POSIX-only, below) can't guard this component
raise OSError(f"Refusing to open lock file through a symlink: {lock_path}")
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | _O_NOFOLLOW, 0o600)
try:
if os.fstat(fd).st_size == 0:
os.write(fd, b"\0") # msvcrt.locking needs at least one byte in the file to lock
os.lseek(fd, 0, 0)
deadline = time.monotonic() + LOCK_WAIT_SECONDS
printed_waiting = False
while True:
try:
_try_lock_nonblocking(fd)
break
except BlockingIOError:
if time.monotonic() >= deadline:
raise LockTimeoutError(
f"Another caveman-compress run appears to be compressing {filepath} "
f"(lock: {lock_path}). Giving up after {LOCK_WAIT_SECONDS}s — retry once "
"it finishes."
) from None
if not printed_waiting: # 900s of silence reads as a hang; tell the user once why nothing's happening yet. flush explicitly since stdout is a pipe when this script runs non-interactively
print(f"Waiting for another caveman-compress run to finish with {filepath}...", flush=True)
printed_waiting = True
time.sleep(LOCK_POLL_INTERVAL)
except OSError as e:
if e.errno in (errno.EOPNOTSUPP, errno.ENOSYS): # filesystem doesn't implement flock/byte-range locking at all (some NFS/SMB/FUSE mounts) — degrade to no coordination rather than fail a run that worked before this lock existed. ENOLCK deliberately excluded: it can mean transient kernel lock-record exhaustion, so it isn't a reliable "unsupported" signal — treating it as one could let a genuinely contended lock proceed unlocked
print(f"⚠️ {lock_dir}'s filesystem doesn't support file locking — proceeding without cross-session coordination.", flush=True)
break
raise
try:
yield
finally:
_unlock(fd)
finally:
os.close(fd)
def is_sensitive_path(filepath: Path) -> bool:
"""Heuristic denylist for files that must never be shipped to a third-party API."""
name = filepath.name
if SENSITIVE_BASENAME_REGEX.match(name):
return True
# Normalize every component, not only basename: directories named
# `api-keys`, `private_keys`, or singular `secret` are equally sensitive.
normalized_parts = {
re.sub(r"[_\-\s.]", "", part.lower()) for part in filepath.parts
}
if normalized_parts & SENSITIVE_PATH_COMPONENTS:
return True
return any(
token in part
for part in normalized_parts
for token in SENSITIVE_NAME_TOKENS
)
def strip_llm_wrapper(text: str) -> str:
r"""Strip an outer ```markdown ... ``` fence when it wraps the ENTIRE output.
The wrapper is only real when the first and last fence lines are the SAME
block. The old regex (``\A\s*(fence)[^\n]*\n(.*)\n\1\s*\Z`` with DOTALL and
a greedy ``.*``) never checked that: it matched any document that merely
STARTS and ENDS with a fence line. An ordinary README section —
```bash npm install``` , prose, ```bash npm test``` — came back with its
first and last fence markers deleted and its two code blocks merged into
prose, so validation failed on both the compress and the fix path and the
section was permanently uncompressible after three paid API calls.
"""
lines = text.split("\n")
first, last = 0, len(lines) - 1
while first < len(lines) and not lines[first].strip():
first += 1
while last > first and not lines[last].strip():
last -= 1
if first >= last:
return text
opener = FENCE_LINE_REGEX.match(lines[first])
closer = FENCE_LINE_REGEX.match(lines[last])
if not opener or not closer:
return text
marker = opener.group(1)
# Closing fence: same character, at least as long, and nothing else on the line.
if closer.group(1)[0] != marker[0] or len(closer.group(1)) < len(marker):
return text
if lines[last].strip() != closer.group(1):
return text
# Any fence of the same kind in between means these two are not one block.
for line in lines[first + 1:last]:
inner = FENCE_LINE_REGEX.match(line)
if inner and inner.group(1)[0] == marker[0] and len(inner.group(1)) >= len(marker):
return text
return "\n".join(lines[first + 1:last])
def write_text_atomic(path: Path, text: str, newline: str = "\n") -> None:
"""Write ``text`` to ``path`` atomically as UTF-8.
Path.write_text() truncates the destination before encoding the string —
a UnicodeEncodeError (or any other failure) partway through leaves a
0-byte file, destroying whatever was there before (issue #655). Encode
first, write the bytes to a sibling temp file, fsync, then os.replace()
so the destination only ever moves from one complete, valid file to
another. Preserves the original file's permission bits across the swap.
``newline`` is the line terminator to emit. Callers pass the terminator
read_source() found in the source file so a CRLF document stays CRLF —
text-mode writes translating LF to the platform default rewrote every
line ending in every file the tool touched (issue #762), and reading the
bytes ourselves means nothing translates them back.
"""
if newline != "\n":
# Normalise first: model output can already carry CRLF, and a bare
# "\n" -> "\r\n" replace would turn those into "\r\r\n".
text = text.replace("\r\n", "\n").replace("\n", newline)
write_bytes_atomic(path, text.encode("utf-8"))
def write_bytes_atomic(path: Path, data: bytes) -> None:
"""Write ``data`` to ``path`` atomically, preserving permission bits."""
fd, tmp_name = tempfile.mkstemp(
dir=str(path.parent), prefix=path.name + ".", suffix=".tmp"
)
tmp_path = Path(tmp_name)
try:
with os.fdopen(fd, "wb") as f:
f.write(data)
f.flush()
os.fsync(f.fileno())
if path.exists():
os.chmod(tmp_path, stat.S_IMODE(path.stat().st_mode))
os.replace(tmp_path, path)
except Exception:
try:
tmp_path.unlink()
except OSError:
pass
raise
def read_source(filepath: Path) -> tuple[str, str, bytes]:
"""Read a source file as UTF-8, returning (text, line_terminator, raw_bytes).
Decodes strictly. The old errors="ignore" silently DROPPED every byte that
was not valid UTF-8 — a cp1252-authored file holding `\xe9` for "e-acute"
lost that byte, the mangled text was what got written to the backup, the
backup readback compared mangled-to-mangled so verification passed, and
then the original was overwritten. The bytes were unrecoverable and
nothing reported a problem (the destructive form of issue #686). A file we
cannot read exactly is a file we must not rewrite.
Line endings are detected from the raw bytes and returned to the caller
rather than being universal-newline'd away, so write_text_atomic can put
back what was there (issue #762). A mixed-ending file takes the terminator
the majority of its lines use — presence of one CRLF is not a mandate to
rewrite every LF in the document. The raw bytes come back too, so the
backup can be a byte-for-byte copy rather than a re-rendering.
"""
raw = filepath.read_bytes()
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as e:
raise ValueError(
f"Refusing to compress {filepath}: not valid UTF-8 "
f"(byte 0x{raw[e.start]:02x} at offset {e.start}). "
"Compression rewrites the file in place, and any byte this tool "
"cannot decode would be destroyed by the round trip. "
"Convert the file to UTF-8 first."
) from None
crlf = text.count("\r\n")
newline = "\r\n" if crlf * 2 > text.count("\n") else "\n"
return text.replace("\r\n", "\n").replace("\r", "\n"), newline, raw
def first_nonblank_line(text: str) -> str:
"""Return the first non-blank line, stripped — used to detect a prose
preamble smuggled in ahead of the real content (issue #588)."""
for line in text.splitlines():
if line.strip():
return line.strip()
return ""
def _write_target(filepath: Path, text: str | bytes, backup_path: Path, newline: str = "\n") -> None:
"""Write to the target file, surfacing the backup location if the write
itself fails. write_text_atomic already leaves the target untouched on
failure, but the caller still needs to know where the pre-compression
original lives instead of being left to guess (issue #652).
``bytes`` restore the source verbatim; ``str`` is model output that still
has to be rendered with the document's line terminator."""
try:
if isinstance(text, bytes):
write_bytes_atomic(filepath, text)
else:
write_text_atomic(filepath, text, newline)
except Exception:
print(f"❌ Write to {filepath} failed. Original preserved at backup: {backup_path}")
raise
from .detect import should_compress
from .validate import validate
MAX_RETRIES = 2
MAX_FILE_SIZE_BYTES = 500_000
MAX_FILE_SIZE_LABEL = "500KB"
ENV_ANTHROPIC_API_KEY = "ANTHROPIC_API_KEY"
ENV_COMPRESS_PROVIDER = "CAVEMAN_COMPRESS_PROVIDER"
ENV_COMPRESS_MODEL = "CAVEMAN_COMPRESS_MODEL"
ENV_FALLBACK_MODEL = "CAVEMAN_MODEL"
ENV_COMPRESS_ENDPOINT = "CAVEMAN_COMPRESS_ENDPOINT"
ENV_COMPRESS_API_KEY = "CAVEMAN_COMPRESS_API_KEY"
PROVIDER_CLAUDE = "claude"
PROVIDER_ANTHROPIC = "anthropic"
PROVIDER_OPENCODE = "opencode"
PROVIDER_OPENAI_COMPAT = "openai-compat"
SUPPORTED_PROVIDERS = frozenset({
PROVIDER_CLAUDE,
PROVIDER_ANTHROPIC,
PROVIDER_OPENCODE,
PROVIDER_OPENAI_COMPAT,
})
DEFAULT_OPENAI_COMPAT_ENDPOINT = "http://localhost:11434/v1" # Ollama
DEFAULT_CLAUDE_MODEL = "claude-sonnet-4-5"
CLAUDE_CLI = "claude"
CLAUDE_PRINT_ARG = "--print"
CLAUDE_SETTING_SOURCES_ARG = "--setting-sources"
CLAUDE_STRICT_MCP_ARG = "--strict-mcp-config"
OPENCODE_CLI = "opencode"
OPENCODE_RUN_ARG = "run"
OPENCODE_STANDALONE_ARG = "--standalone"
OPENCODE_FILE_ARG = "--file"
OPENCODE_AGENT_ARG = "--agent"
OPENCODE_AGENT = "caveman-compress"
MODEL_ARG = "--model"
OPENCODE_PROMPT_PREFIX = ".caveman-compress-prompt-"
OPENCODE_PROMPT_SUFFIX = ".md"
OPENCODE_PROMPT_MESSAGE = "Follow the attached prompt exactly. Return only the final answer."
# The compressed file is untrusted input sent as a prompt to opencode's agent.
# opencode 2.x allows every action not explicitly denied (bash, edit,
# websearch, MCP tools, subagents), so deny all of them: compression needs no
# tools. Probed on 2.0.22 --standalone: no tools are offered to the model and
# the --file attachment is still read. A per-agent rule is applied after the
# global one and the last match wins, so the user's own
# `agent.build.permission` could re-allow a tool for the default agent; the
# deny is therefore also bound to OPENCODE_AGENT, an agent only we define.
OPENCODE_PERMISSION_DENY = {"*": "deny"}
# Output ceiling for one SDK call: the max output of the default model
# (claude-sonnet-4-5). A body that needs more is refused in call_claude(),
# never written back truncated.
MAX_OUTPUT_TOKENS = 64000
def resolve_max_output_tokens() -> int:
"""Output ceiling for one SDK call, overridable per model.
The default tracks the max output of the default model. CAVEMAN_MODEL lets
a user pin a model whose ceiling is lower (claude-opus-4-0 tops out at
32K), and asking for more output than the model allows is a 400 from the
API — an opaque failure in place of the truncation this cap exists to
prevent. CAVEMAN_MAX_OUTPUT_TOKENS lets that user lower the ask to match.
A malformed value raises rather than silently falling back: an override
that is quietly ignored looks like the cap was raised when it was not, and
the symptom is the same truncation, one layer further from the cause.
"""
raw = os.environ.get("CAVEMAN_MAX_OUTPUT_TOKENS", "").strip()
if not raw:
return MAX_OUTPUT_TOKENS
try:
value = int(raw)
except ValueError:
raise RuntimeError(
f"CAVEMAN_MAX_OUTPUT_TOKENS must be a positive integer, got {raw!r}."
) from None
if value < 1:
raise RuntimeError(
f"CAVEMAN_MAX_OUTPUT_TOKENS must be a positive integer, got {raw!r}."
)
return value
def _is_smaller_than_body(candidate_body: str, body: str) -> bool:
"""True when `candidate_body` actually compresses `body`.
The non-expansion invariant for #776. It lives in a helper because it has
to hold for EVERY candidate, not just the first one: a candidate that fails
validation is sent back to the provider for repair, and the repaired text is what
gets written if it validates. Checking only the first candidate left the
retry path able to write a longer file and report it as a successful
compression — the original bug, one branch over.
Always compares bodies with frontmatter already removed. Frontmatter is
preserved verbatim, so counting it on one side and not the other would
measure the wrong thing.
"""
candidate_len = len(candidate_body.strip())
body_len = len(body.strip())
if candidate_len >= body_len:
print(
"❌ Compression aborted: output is not smaller than input "
f"({candidate_len} >= {body_len} chars)."
)
return False
return True
# Bounds each individual Claude call so a stalled CLI (dropped network, an
# auth prompt with no TTY to answer it) can't hang past what LOCK_WAIT_SECONDS
# assumes for the whole run's worst case (MAX_RETRIES+1 calls).
CLAUDE_CALL_TIMEOUT_SECONDS = LOCK_WAIT_SECONDS // (MAX_RETRIES + 1)
# ---------- LLM Calls ----------
class AnthropicSdkUnavailable(RuntimeError):
"""Raised when the explicitly selected Anthropic SDK is unavailable."""
def configured_provider() -> str:
provider = (
os.environ.get(ENV_COMPRESS_PROVIDER) or PROVIDER_CLAUDE
).strip().lower()
if provider not in SUPPORTED_PROVIDERS:
supported = ", ".join(sorted(SUPPORTED_PROVIDERS))
raise ValueError(
f"Unsupported caveman-compress provider: {provider}. "
f"Supported providers: {supported}"
)
return provider
def configured_model(default_model: Optional[str] = None) -> Optional[str]:
model = (
os.environ.get(ENV_COMPRESS_MODEL)
or os.environ.get(ENV_FALLBACK_MODEL)
or default_model
)
if model is None:
return None
return model.strip() or None
def run_cli(
binary_name: str,
args: List[str],
stdin_prompt: Optional[str] = None,
extra_env: Optional[dict] = None,
) -> str:
binary = shutil.which(binary_name) or binary_name
command = [binary, *args]
run_kwargs = {
"text": True,
"capture_output": True,
"check": True,
"encoding": "utf-8",
"errors": "replace",
"timeout": CLAUDE_CALL_TIMEOUT_SECONDS,
}
if stdin_prompt is not None:
run_kwargs["input"] = stdin_prompt
if extra_env:
# The child inherits this process's environment, plus extra_env.
run_kwargs["env"] = {**environ, **extra_env}
try:
result = subprocess.run(command, **run_kwargs)
return strip_llm_wrapper(result.stdout.strip())
except FileNotFoundError as error:
raise RuntimeError(f"{binary_name} CLI not found on PATH") from error
except subprocess.CalledProcessError as error:
stderr = error.stderr.strip() if error.stderr else "no stderr"
raise RuntimeError(f"{binary_name} call failed:\n{stderr}") from error
except subprocess.TimeoutExpired as error:
raise RuntimeError(
f"{binary_name} CLI call timed out after "
f"{CLAUDE_CALL_TIMEOUT_SECONDS}s "
"(stalled network, or an auth prompt with no TTY to answer it)"
) from error
def call_anthropic_sdk(prompt: str) -> str:
api_key = os.environ.get(ENV_ANTHROPIC_API_KEY)
if not api_key:
raise RuntimeError(
f"{ENV_ANTHROPIC_API_KEY} is required for provider "
f"'{PROVIDER_ANTHROPIC}'"
)
try:
import anthropic
except ImportError as error:
raise AnthropicSdkUnavailable(
f"anthropic package is required for provider '{PROVIDER_ANTHROPIC}'"
) from error
max_output_tokens = resolve_max_output_tokens()
client = anthropic.Anthropic(
api_key=api_key,
timeout=CLAUDE_CALL_TIMEOUT_SECONDS,
)
# Streaming, not create(): a large compression needs far more than
# the old 8192-token cap, and tokens that arrive as they are
# generated keep a long call from tripping the request timeout.
with client.messages.stream(
model=configured_model(DEFAULT_CLAUDE_MODEL),
max_tokens=max_output_tokens,
messages=[{"role": "user", "content": prompt}],
) as stream:
message = stream.get_final_message()
# At the cap the tail was never generated. validate() and the fix
# prompt cannot recover it, so fail now instead of paying for
# retries that end by restoring the original anyway.
if message.stop_reason == "max_tokens":
raise RuntimeError(
f"Claude output hit the {max_output_tokens}-token cap, so the result is incomplete. "
"Split the file into smaller parts and compress each one."
)
# Tool-heavy models can put a tool_use or thinking block first.
text = next(
(
block.text
for block in message.content
if getattr(block, "type", None) == "text"
),
"",
)
return strip_llm_wrapper(text.strip())
def call_claude_cli(prompt: str) -> str:
args = []
model = configured_model()
if model:
args.extend([MODEL_ARG, model])
args.extend([
CLAUDE_PRINT_ARG,
CLAUDE_SETTING_SOURCES_ARG,
"",
CLAUDE_STRICT_MCP_ARG,
])
return run_cli(CLAUDE_CLI, args, prompt)
def call_opencode_cli(prompt: str) -> str:
args = [OPENCODE_RUN_ARG]
model = configured_model()
if model:
args.extend([MODEL_ARG, model])
args.extend([OPENCODE_AGENT_ARG, OPENCODE_AGENT])
# --standalone: the shared background service ignores this process's env
# (verified on opencode 2.0.22), so the deny config only binds a private
# server. Never pass --auto. Any inline config the user set is kept.
args.append(OPENCODE_STANDALONE_ARG)
try:
config = json.loads(os.environ.get("OPENCODE_CONFIG_CONTENT") or "{}")
except ValueError:
config = None
if not isinstance(config, dict):
raise RuntimeError(
"OPENCODE_CONFIG_CONTENT must be a strict JSON object (no comments or "
"trailing commas) so compress can add its tool-deny rule; fix or unset it."
)
agents = config.get("agent") or {}
if not isinstance(agents, dict):
raise RuntimeError(
'OPENCODE_CONFIG_CONTENT "agent" must be a JSON object so compress '
"can add its tool-deny agent; fix or unset it."
)
config["permission"] = OPENCODE_PERMISSION_DENY
config["agent"] = {
**agents,
OPENCODE_AGENT: {"permission": OPENCODE_PERMISSION_DENY},
}
extra_env = {"OPENCODE_CONFIG_CONTENT": json.dumps(config)}
prompt_path = None
try:
with tempfile.NamedTemporaryFile(
"w",
encoding="utf-8",
prefix=OPENCODE_PROMPT_PREFIX,
suffix=OPENCODE_PROMPT_SUFFIX,
delete=False,
) as prompt_file:
prompt_path = Path(prompt_file.name)
prompt_file.write(prompt)
args.extend(
[OPENCODE_FILE_ARG, str(prompt_path), OPENCODE_PROMPT_MESSAGE]
)
return run_cli(OPENCODE_CLI, args, extra_env=extra_env)
finally:
if prompt_path is not None:
# Warn, never raise: raising here would replace the real opencode
# error (or a good result) with a cleanup failure.
try:
prompt_path.unlink(missing_ok=True)
except OSError:
print(
f"warning: could not delete temporary opencode prompt {prompt_path}",
file=sys.stderr,
)
def call_openai_compat(prompt: str) -> str:
"""POST to any OpenAI-compatible /chat/completions (Ollama, llama.cpp,
vLLM, LM Studio). Standard library only, so no new dependency."""
model = configured_model()
if not model:
raise RuntimeError(
f"{ENV_COMPRESS_MODEL} is required for provider '{PROVIDER_OPENAI_COMPAT}'"
)
base = os.environ.get(ENV_COMPRESS_ENDPOINT, DEFAULT_OPENAI_COMPAT_ENDPOINT).rstrip("/")
body = {"model": model, "messages": [{"role": "user", "content": prompt}], "stream": False}
request = urllib.request.Request(
f"{base}/chat/completions",
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
api_key = os.environ.get(ENV_COMPRESS_API_KEY)
if api_key:
# Unredirected: urllib copies ordinary headers onto a redirect, even
# one to another host.
request.add_unredirected_header("Authorization", f"Bearer {api_key}")
try:
with urllib.request.urlopen(request, timeout=CLAUDE_CALL_TIMEOUT_SECONDS) as response:
raw = response.read()
except urllib.error.HTTPError as error:
detail = error.read().decode("utf-8", "replace")[:500]
raise RuntimeError(
f"{PROVIDER_OPENAI_COMPAT} call failed: HTTP {error.code}: {detail}"
) from error
except OSError as error: # URLError, refused connection, timeout
reason = getattr(error, "reason", error)
raise RuntimeError(f"{PROVIDER_OPENAI_COMPAT} call to {base} failed: {reason}") from error
try:
choice = json.loads(raw)["choices"][0]
content = choice["message"].get("content")
except (ValueError, KeyError, IndexError, TypeError, AttributeError):
detail = raw.decode("utf-8", "replace")[:500]
raise RuntimeError(
f"{PROVIDER_OPENAI_COMPAT} call to {base} returned an unexpected response: {detail}"
) from None
# Same rule as the SDK path: a truncated body is never returned.
if choice.get("finish_reason") == "length":
raise RuntimeError(
f"{PROVIDER_OPENAI_COMPAT} output hit the server's token cap, so the result is "
"incomplete. Split the file into smaller parts and compress each one."
)
return strip_llm_wrapper((content or "").strip())
def call_claude(prompt: str) -> str:
"""Send a prompt to the configured compression provider.
Prefers the Anthropic SDK when ANTHROPIC_API_KEY is set; otherwise falls
back to the ``claude --print`` CLI (which handles desktop auth). Set
``CAVEMAN_COMPRESS_PROVIDER=opencode`` and ``CAVEMAN_COMPRESS_MODEL`` (or
``CAVEMAN_MODEL``) to route compression through opencode instead, or
``openai-compat`` for an OpenAI-compatible server such as local Ollama.
On Windows the CLI subprocess decoding defaults to the system codepage
(cp1251 / cp1252) and crashes on UTF-8 output — see issue #152. Pinning
``encoding="utf-8"`` with ``errors="replace"`` matches the CLI's actual
native I/O and prevents the UnicodeDecodeError before validation can
report. Windows users with non-ASCII content can also set
``ANTHROPIC_API_KEY`` to route through the SDK and skip the subprocess.
"""
provider = configured_provider()
if provider == PROVIDER_OPENCODE:
return call_opencode_cli(prompt)
if provider == PROVIDER_OPENAI_COMPAT:
return call_openai_compat(prompt)
if provider == PROVIDER_ANTHROPIC:
return call_anthropic_sdk(prompt)
if os.environ.get(ENV_ANTHROPIC_API_KEY):
try:
return call_anthropic_sdk(prompt)
except AnthropicSdkUnavailable:
pass
return call_claude_cli(prompt)
def build_compress_prompt(original: str) -> str:
return f"""
Compress this markdown into caveman format.
STRICT RULES:
- Do NOT modify anything inside ``` code blocks
- Do NOT modify anything inside a 4-space-indented code block either — those are code too, and they are validated
- Do NOT modify anything inside inline backticks
- Do NOT add new ``` fences around content that is not already fenced, even if it looks like code or JSON
- Do NOT create, remove, reword or promote headings; keep existing # headings exactly
- Do NOT convert XML-like tags (e.g. <example>, <section>) into headings or other markdown; leave them as plain text
- Preserve ALL URLs exactly
- Preserve ALL headings exactly
- Preserve file paths and commands
- Return ONLY the compressed markdown body, with no preface, commentary, or YAML frontmatter — do NOT wrap the entire output in a ```markdown fence or any other fence. Inner code blocks from the original stay as-is; do not add a new outer fence around the whole file.
- Output ends where the input ends: no appended summary, notes, "untouched" section, or trailing commentary, and no new --- or *** break to introduce one
- Keep the verb that carries each instruction: "Do X", "Run Y", "Never Z" stay orders. A fragment is fine, a verbless noun phrase is not
Only compress natural language prose. Never restructure, add formatting, or improve organization.
TEXT:
{original}
"""
def build_fix_prompt(original: str, compressed: str, errors: List[str]) -> str:
errors_str = "\n".join(f"- {e}" for e in errors)
return f"""You are fixing a caveman-compressed markdown file. Specific validation errors were found.
CRITICAL RULES:
- DO NOT recompress or rephrase the file
- ONLY fix the listed errors — leave everything else exactly as-is
- The ORIGINAL is provided as reference only (to restore missing content)
- Preserve caveman style in all untouched sections
ERRORS TO FIX:
{errors_str}
HOW TO FIX:
- Missing URL: find it in ORIGINAL, restore it exactly where it belongs in COMPRESSED
- Code block mismatch: find the exact code block in ORIGINAL, restore it in COMPRESSED
- Heading mismatch: restore the exact heading text from ORIGINAL into COMPRESSED
- Do not touch any section not mentioned in the errors
ORIGINAL (reference only):
{original}
COMPRESSED (fix this):
{compressed}
Return ONLY the fixed compressed markdown body. No preface, commentary, YAML frontmatter, or explanation.
"""
CODE_MARKER_PREFIX = "@@CAVEMAN_PRESERVED_CODE_"
FENCE_OPEN_RE = re.compile(r"^[ ]{0,3}(`{3,}|~{3,})(?:[^\r\n]*)$")
# Whole-line tags only, at most 3 spaces in (4+ is an indented code block).
NOCOMPRESS_OPEN_RE = re.compile(r"^[ ]{0,3}<!--\s*nocompress\s*-->\s*$", re.IGNORECASE)
NOCOMPRESS_CLOSE_RE = re.compile(r"^[ ]{0,3}<!--\s*/nocompress\s*-->\s*$", re.IGNORECASE)
# CommonMark HTML block type 2: a line opening with `<!--` (up to 3 spaces in)
# runs to the first line holding `-->`. Comments often carry notes for the next
# editor; unmasked, the model could drop or reword them and validation, which
# never counted them, passed the result (#1350).
HTML_COMMENT_OPEN_RE = re.compile(r"^[ ]{0,3}<!--")
def mask_code_blocks(text: str) -> Tuple[str, List[Tuple[str, str]]]:
"""Replace fenced and four-space-indented code, <!-- nocompress -->
regions and block-level HTML comments with opaque line markers."""
if CODE_MARKER_PREFIX in text:
raise ValueError("Input contains reserved Caveman code-preservation marker")
lines = text.splitlines(keepends=True)
out: List[str] = []
blocks: List[Tuple[str, str]] = []
i = 0
while i < len(lines):
line_without_newline = lines[i].rstrip("\r\n")
nocompress = NOCOMPRESS_OPEN_RE.match(line_without_newline)
fence = not nocompress and FENCE_OPEN_RE.match(line_without_newline)
indented = bool(line_without_newline) and (
line_without_newline.startswith(" ") or line_without_newline.startswith("\t")
)
comment = (
not nocompress and not fence and not indented
and HTML_COMMENT_OPEN_RE.match(line_without_newline)
)
if not nocompress and not fence and not indented and not comment:
out.append(lines[i])
i += 1
continue
start = i
if comment:
# Ends on the first line holding `-->`, which may be this one. An
# unclosed comment runs to the end of the document, as it renders.
after_open = line_without_newline[line_without_newline.index("<!--") + 4:]
i += 1
if "-->" not in after_open:
while i < len(lines) and "-->" not in lines[i]:
i += 1
i = min(i + 1, len(lines))
elif nocompress:
i += 1
while i < len(lines) and not NOCOMPRESS_CLOSE_RE.match(lines[i].rstrip("\r\n")):
i += 1
if i == len(lines):
raise ValueError(f"unclosed <!-- nocompress --> region starting at line {start + 1}")
i += 1
elif fence:
fence_run = fence.group(1)
close_re = re.compile(
rf"^[ ]{{0,3}}{re.escape(fence_run[0])}{{{len(fence_run)},}}[ \t]*$"
)
i += 1
while i < len(lines):
if close_re.match(lines[i].rstrip("\r\n")):
i += 1
break
i += 1
else:
i += 1
while i < len(lines):
candidate = lines[i].rstrip("\r\n")
if not candidate or candidate.startswith(" ") or candidate.startswith("\t"):
i += 1
continue
break
block = "".join(lines[start:i])
marker = f"{CODE_MARKER_PREFIX}{len(blocks)}_{hashlib.sha256(block.encode('utf-8')).hexdigest()[:16]}@@"
blocks.append((marker, block))
newline = "\r\n" if block.endswith("\r\n") else "\n" if block.endswith("\n") else ""
out.append(marker + newline)
return "".join(out), blocks
def restore_code_blocks(text: str, blocks: List[Tuple[str, str]]) -> str:
"""Restore markers exactly; fail closed if model removed, copied, or altered one."""
restored = text
for marker, block in blocks:
if restored.count(marker) != 1:
raise ValueError(
f"Provider changed preserved code marker {marker}; refusing to write"
)
# Masking gives marker its own transport newline. Consume that wrapper
# when present so restoring a block that already ended in newline does
# not silently add another blank line.
if marker + "\r\n" in restored:
restored = restored.replace(marker + "\r\n", block, 1)
elif marker + "\n" in restored:
restored = restored.replace(marker + "\n", block, 1)
else:
restored = restored.replace(marker, block, 1)
if CODE_MARKER_PREFIX in restored:
raise ValueError("Provider returned an unknown Caveman code-preservation marker")
return restored
def _frontmatter_block_count(masked_body: str) -> int:
"""Count mapping-shaped delimiter blocks outside preserved code."""
lines = masked_body.splitlines()
delimiters = [i for i, line in enumerate(lines) if line == "---"]
count = 0
for start, end in zip(delimiters, delimiters[1:]):
content = [line for line in lines[start + 1:end] if line.strip()]
key_value = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*:(\s.*)?$")
if content and key_value.match(content[0]) and all(
key_value.match(line) or line[0].isspace() or line.startswith("- ")
for line in content[1:]
):
count += 1
return count
_THEMATIC_BREAK_RE = re.compile(r"^(?:-{3,}|\*{3,}|_{3,})$")
def _thematic_break_count(masked_body: str) -> int:
"""Count bare thematic-break lines outside preserved code.
The head of the body is anchored by its first line, but nothing anchored
the tail, so an appended commentary section ("Untouched: ... Risk: ...")
under a new ``---`` was accepted and written into the file (#1433). The
model is already told not to restructure or add formatting, so a break
that was not in the original is an addition either way: counting them
needs no classification of what follows, which is the part #1359
deliberately declines to do.
"""
return sum(
1 for line in masked_body.splitlines()
if _THEMATIC_BREAK_RE.match(line.strip())
)
def _body_candidate_error(original_masked: str, candidate_masked: str) -> str:
"""Use the same body-only structural guards for compression and repair."""
anchor = first_nonblank_line(original_masked)
if anchor.startswith(("#", CODE_MARKER_PREFIX)) and first_nonblank_line(candidate_masked) != anchor:
return "output does not start with the original body's first line (possible preamble leak)"
if _frontmatter_block_count(candidate_masked) > _frontmatter_block_count(original_masked):
return "output added YAML frontmatter to the markdown body"
if _thematic_break_count(candidate_masked) > _thematic_break_count(original_masked):
return (
"output added a thematic break the original did not have "
"(possible appended commentary or notes section)"
)
return ""
# ---------- Core Logic ----------
def compress_file(filepath: Path) -> bool:
# Resolve first so the lock and every check below key off the same canonical path regardless of how the caller spelled it.
filepath = filepath.resolve()
# None of these three checks depends on mutual exclusion, so they run before the lock is taken — a rejected input (bad path, oversized, sensitive name) shouldn't leave a permanent lock file behind in shared state.
if not filepath.exists():
raise FileNotFoundError(f"File not found: {filepath}")
if filepath.stat().st_size > MAX_FILE_SIZE_BYTES:
raise ValueError(
f"File too large to compress safely (max {MAX_FILE_SIZE_LABEL}): "
f"{filepath}"
)
# Refuse files that look like they contain secrets or PII. Compressing ships
# the raw bytes to the configured provider — a third-party boundary — so we fail
# loudly rather than silently exfiltrate credentials or keys. Override is
# intentional: the user must rename the file if the heuristic is wrong.
if is_sensitive_path(filepath):
raise ValueError(
f"Refusing to compress {filepath}: filename looks sensitive "
"(credentials, keys, secrets, or known private paths). "
"Compression sends file contents to the configured LLM provider. "
"Rename the file if this is a false positive."
)
with file_lock(filepath):
backup_path = backup_dir_for(filepath) / (filepath.stem + ".original.md")
had_backup = backup_path.exists()
run = {"live_write_started": False}
try:
return _compress_file_locked(filepath, run)
except BaseException:
# A provider error, timeout or Ctrl-C mid-run must not leave the
# staging file next to the source, or a backup that makes every
# later run abort with "Backup file already exists". The backup is
# removed only if this run created it and never started the live
# write; past that point it may be the only copy of the original.
filepath.with_name(filepath.name + ".caveman-staged").unlink(missing_ok=True)
if not had_backup and not run["live_write_started"]:
backup_path.unlink(missing_ok=True)
raise
def _compress_file_locked(filepath: Path, run: dict) -> bool:
"""Body of compress_file; runs entirely under compress_file's file_lock for this resolved path."""
print(f"Processing: {filepath}")
if not should_compress(filepath):
print("Skipping (not natural language)")
return False
original_text, newline, original_raw = read_source(filepath)
# Store backup outside the source directory so skill auto-loaders don't
# re-ingest the `.original.md` copy as a live file. Mirror the source's
# parent-dir name + stem under a platform-aware base to reduce collisions.
backup_dir = backup_dir_for(filepath)
backup_path = backup_dir / (filepath.stem + ".original.md")
if not original_text.strip():
print("❌ Refusing to compress: file is empty or whitespace-only.")
return False
credential = find_credential(original_text)
if credential is not None:
kind, lineno = credential
raise ValueError(
f"Refusing to compress {filepath}: line {lineno} looks like a {kind}. "
"Compression sends the file's text to the configured LLM provider. "
"Move the secret out of this file (an env var or secret manager), then rerun."
)
# Refuse when a backup already exists, so a rerun cannot capture this file's
# already-compressed text as the new "original".
if backup_path.exists():
print(f"⚠️ Backup already exists: {backup_path}")
print(f" It holds the uncompressed original of an earlier run on {filepath.name}.")
print("Aborting to prevent data loss.")
print(f" To restore it: cp '{backup_path}' '{filepath}'")
print(f" To start over: mv '{backup_path}' '{backup_path}.kept'")
print(" Do not just delete it — rerunning would then store the ALREADY")
print(" COMPRESSED text as the new original, losing the uncompressed version.")
return False
# The pre-#1351 location is checked too: an install that compressed this file
# before the backup key changed still has its only original there. That path
# was keyed on the parent directory's NAME alone, so it was shared by every
# same-named file on the machine — this backup may well belong to a DIFFERENT
# project, which is exactly the bug that produced it. So: refuse, name the
# ambiguity, and never read, move or delete it on the user's behalf.
legacy_backup = legacy_backup_dir_for(filepath) / (filepath.stem + ".original.md")
if legacy_backup.exists():
print(f"⚠️ A backup from an older caveman-compress exists: {legacy_backup}")
print(" Older versions keyed backups on the parent directory's name only,")
print(f" so that file is the original of SOME {filepath.parent.name}/{filepath.name}")
print(" on this machine — possibly this one, possibly another project's.")
print("Aborting to prevent data loss.")
print(" Check its contents, then move it out of the way to proceed:")
print(f" mv '{legacy_backup}' '{legacy_backup}.kept'")
print(" Backups written from now on are keyed on the file's full path and")
print(f" will land in {backup_path.parent}")
return False
# Split YAML frontmatter off before compression. LLMs tend to strip or
# rewrite frontmatter despite preserve-structure rules; we keep it verbatim
# by removing it from the input and re-prepending it to the output.
frontmatter, body = split_frontmatter(original_text)
if frontmatter:
print(f"Detected YAML frontmatter ({len(frontmatter)} chars) — preserving verbatim")
if not body.strip():
print("❌ Refusing to compress: body is empty after frontmatter removal.")
return False
# Step 1: Compress (body only, frontmatter excluded)
provider = configured_provider()
print(f"Compressing with {provider}...")
masked_body, code_blocks = mask_code_blocks(body)
masked_compressed = call_claude(build_compress_prompt(masked_body))
if masked_compressed is None or not masked_compressed.strip():
print(f"❌ Compression aborted: {provider} returned an empty response.")
print(" Original file is untouched (no backup created).")
return False
body_error = _body_candidate_error(masked_body, masked_compressed)
if body_error:
print(f"❌ Compression aborted: {body_error}.")
print(" Original file is untouched (no backup created).")
return False
try:
compressed_body = restore_code_blocks(masked_compressed, code_blocks)
except ValueError as error:
print(f"❌ Compression aborted: {error}")
print(" Original file is untouched (no backup created).")
return False
if compressed_body is None or not compressed_body.strip():
print(f"❌ Compression aborted: {provider} returned an empty response.")
print(" Original file is untouched (no backup created).")
return False
# Compare the BODY (not the whole file) — frontmatter is preserved verbatim
# and would never change, so identity must be judged on the compressible part.
if compressed_body.strip() == body.strip():
print("❌ Compression aborted: output is identical to input.")
print(f" Likely causes: {provider} refused, returned the prompt verbatim, or the file is")
print(" already in caveman form. Original file is untouched (no backup created).")
return False
# A rewrite that is structurally faithful but LONGER than the input passes
# every check below (validate() only checks structural invariants, not
# length) and would otherwise be written over the original and reported
# as a successful compression — the opposite of what this tool exists to
# do (issue #776). Same length is also a reject: a compression that saved
# nothing isn't a compression.
if not _is_smaller_than_body(compressed_body, body):
print(" Original file is untouched (no backup created).")
return False
# Reassemble: frontmatter (verbatim) + compressed body
compressed = frontmatter + compressed_body
# Save original as backup, then verify the backup readback before
# touching the input file. If the filesystem dropped bytes (encoding,
# antivirus, disk full), unlink the bad backup and abort instead of
# leaving the user with a corrupt backup + compressed primary.
backup_dir.mkdir(parents=True, exist_ok=True)
write_bytes_atomic(backup_path, original_raw)
if backup_path.read_bytes() != original_raw:
print(f"❌ Backup write verification failed: {backup_path}")
print(" In-memory original differs from on-disk backup. Aborting before touching the input file.")
try:
backup_path.unlink()
except OSError:
pass
return False
# Step 2: Validate + Retry. Each candidate is staged and validated next
# to the source; the live file is written only once one passes (#544).
staging_path = filepath.with_name(filepath.name + ".caveman-staged")
for attempt in range(MAX_RETRIES):
print(f"\nValidation attempt {attempt + 1}")
_write_target(staging_path, compressed, backup_path, newline)
result = validate(backup_path, staging_path)
if result.is_valid:
print("Validation passed")
# Warnings don't block the write, but they were computed and never
# shown (#1350): a big bullet-count drop is worth a second look.
for warning in result.warnings:
print(f" ⚠️ {warning}")
run["live_write_started"] = True
_write_target(filepath, compressed, backup_path, newline)
staging_path.unlink(missing_ok=True)
return True
print("❌ Validation failed:")
for err in result.errors:
print(f" - {err}")
if attempt == MAX_RETRIES - 1:
staging_path.unlink(missing_ok=True)
backup_path.unlink(missing_ok=True)
print("Failed after retries: original left untouched")
return False
print(f"Fixing with {provider}...")
fixed = call_claude(
build_fix_prompt(body, compressed_body, result.errors)
)
if fixed is None or not fixed.strip():
print(f"❌ Fix attempt aborted: {provider} returned an empty response.")
print(" Skipping this attempt.")
continue
# Repairs use the same body-only contract as the initial compression.
# Mask code for structural checks so fenced YAML is never frontmatter.
try:
masked_fixed, _ = mask_code_blocks(fixed)
except ValueError as error:
print(f"❌ Fix attempt aborted: {error}")
continue
body_error = _body_candidate_error(masked_body, masked_fixed)
if body_error:
print(f"❌ Fix attempt aborted: {body_error}.")
print(" Skipping this attempt.")
continue
fixed_body = fixed
if not _is_smaller_than_body(fixed_body, body):
print(" Skipping this attempt.")
continue
# The fix prompt carries the body unmasked; every protected block
# must still come back verbatim, including list-nested fences.
if any(block.rstrip("\n") not in fixed_body for _, block in code_blocks):
print("❌ Fix attempt aborted: output changed a preserved code or <!-- nocompress --> block.")
print(" Skipping this attempt.")
continue
compressed_body = fixed_body
compressed = frontmatter + compressed_body
return False
scripts/detect.py
#!/usr/bin/env python3
"""Detect whether a file is natural language (compressible) or code/config (skip)."""
import json
import re
from pathlib import Path
# Extensions that are natural language and compressible. `.mdc` is a Cursor
# rule: prose with YAML frontmatter, the same shape as a SKILL.md, and the
# format caveman's own `--with-init` writes to `.cursor/rules/caveman.mdc`.
# Content sniffing below runs only for EXTENSIONLESS files, so an extension
# that is on neither list falls through to `unknown` and is never compressed.
COMPRESSIBLE_EXTENSIONS = {".md", ".mdc", ".txt", ".markdown", ".rst", ".typ", ".typst", ".tex"}
# Extensions that are code/config and should be skipped
SKIP_EXTENSIONS = {
".py", ".js", ".ts", ".tsx", ".jsx", ".json", ".yaml", ".yml",
".toml", ".env", ".lock", ".css", ".scss", ".html", ".xml",
".sql", ".sh", ".bash", ".zsh", ".go", ".rs", ".java", ".c",
".cpp", ".h", ".hpp", ".rb", ".php", ".swift", ".kt", ".lua",
".dockerfile", ".makefile", ".csv", ".ini", ".cfg",
}
# Well-known build/config files that carry no (or a misleading) extension —
# `Dockerfile` has no suffix so `.dockerfile` above never matches it, and
# `CMakeLists.txt` would ride the compressible `.txt` rule. Checked by
# basename before any extension rule.
KNOWN_CODE_FILENAMES = {
"dockerfile", "containerfile", "makefile", "gnumakefile", "jenkinsfile",
"vagrantfile", "rakefile", "gemfile", "justfile", "procfile", "brewfile",
"earthfile", "fastfile", "podfile",
"cmakelists.txt",
}
# Dotfiles whose entire name sits where pathlib expects a suffix, so
# `Path(".cursorrules").suffix` is `""` just as it is for `.env`. These are the
# prose ones — Cursor's and its imitators' legacy single-file rule formats —
# and the only bare dotfiles that may be compressed. Everything else is
# treated as config; see the dotfile branch in `detect_file_type`.
PROSE_DOTFILES = {
".cursorrules", ".windsurfrules", ".clinerules", ".aiderrules",
".goosehints", ".augmentrules",
}
# Extensions on SKIP_EXTENSIONS that are data/config rather than program text.
CONFIG_EXTENSIONS = {".json", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".env"}
# Patterns that indicate a line is code
CODE_PATTERNS = [
re.compile(r"^\s*(import |from .+ import |require\(|const |let |var )"),
re.compile(r"^\s*(def |class |function |async function |export )"),
re.compile(r"^\s*(if\s*\(|for\s*\(|while\s*\(|switch\s*\(|try\s*\{)"),
re.compile(r"^\s*[\}\]\);]+\s*$"), # closing braces/brackets
re.compile(r"^\s*@\w+"), # decorators/annotations
re.compile(r'^\s*"[^"]+"\s*:\s*'), # JSON-like key-value
re.compile(r"^\s*\w+\s*=\s*[{\[\(\"']"), # assignment with literal
]
def _is_code_line(line: str) -> bool:
"""Check if a line looks like code."""
return any(p.match(line) for p in CODE_PATTERNS)
def _is_json_content(text: str) -> bool:
"""Check if content is valid JSON."""
try:
json.loads(text)
return True
except (json.JSONDecodeError, ValueError):
return False
def _is_yaml_content(lines: list[str]) -> bool:
"""Heuristic: check if content looks like YAML."""
yaml_indicators = 0
for line in lines[:30]:
stripped = line.strip()
if stripped.startswith("---"):
yaml_indicators += 1
elif re.match(r"^\w[\w\s]*:\s", stripped):
yaml_indicators += 1
elif stripped.startswith("- ") and ":" in stripped:
yaml_indicators += 1
# If most non-empty lines look like YAML
non_empty = sum(1 for l in lines[:30] if l.strip())
return non_empty > 0 and yaml_indicators / non_empty > 0.6
def detect_file_type(filepath: Path) -> str:
"""Classify a file as 'natural_language', 'code', 'config', or 'unknown'.
Returns:
One of: 'natural_language', 'code', 'config', 'unknown'
"""
ext = filepath.suffix.lower()
name = filepath.name.lower()
# Known code filenames win over any extension rule
if name in KNOWN_CODE_FILENAMES:
return "code"
# A bare dotfile carries its whole name where pathlib expects a suffix, so
# `Path(".env").suffix` is `""` and every table below was skipped. The file
# then reached the extensionless content sniffer, which only recognises
# brace-and-quote code — so `.env`, `.gitignore` and `.editorconfig` all
# came back `natural_language` and `/caveman-compress .gitignore` shipped
# the bytes to the configured provider and rewrote the file in place.
# Resolve the name against the tables, and treat anything unrecognised as
# config rather than prose: a dotfile is an rc/config file by convention,
# and guessing wrong is destructive in both directions at once. Guessing
# "config" only costs a `Skipping` line on a file that is left untouched.
if not ext and name.startswith(".") and name not in {".", ".."}:
if name in PROSE_DOTFILES or name in COMPRESSIBLE_EXTENSIONS:
return "natural_language"
if name in SKIP_EXTENSIONS:
ext = name # fall through to the shared config/code split below
else:
return "config"
# Extension-based classification
if ext in COMPRESSIBLE_EXTENSIONS:
return "natural_language"
if ext in SKIP_EXTENSIONS:
return "code" if ext not in CONFIG_EXTENSIONS else "config"
# Extensionless files (like CLAUDE.md, TODO) — check content
if not ext:
try:
text = filepath.read_text(encoding="utf-8", errors="ignore")
except (OSError, PermissionError):
return "unknown"
lines = text.splitlines()[:50]
# Shebang means executable script, never prose
if text.startswith("#!"):
return "code"
if _is_json_content(text[:10000]):
return "config"
if _is_yaml_content(lines):
return "config"
code_lines = sum(1 for l in lines if l.strip() and _is_code_line(l))
non_empty = sum(1 for l in lines if l.strip())
if non_empty > 0 and code_lines / non_empty > 0.4:
return "code"
return "natural_language"
return "unknown"
def should_compress(filepath: Path) -> bool:
"""Return True if the file is natural language and should be compressed."""
if not filepath.is_file():
return False
# Skip backup files
if filepath.name.endswith(".original.md"):
return False
if is_generated(filepath):
return False
return detect_file_type(filepath) == "natural_language"
GENERATED_MARKER_RE = re.compile(r"\bgenerated by\b|@generated\b", re.IGNORECASE)
def is_generated(filepath: Path) -> bool:
"""True when line 1 says the file is generated. Its generator rewrites it,
so compressing it only spends tokens on text about to be replaced (#1350)."""
try:
with filepath.open(encoding="utf-8", errors="replace") as handle:
first_line = handle.readline(512)
except OSError:
return False
return bool(GENERATED_MARKER_RE.search(first_line))
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python detect.py <file1> [file2] ...")
sys.exit(1)
for path_str in sys.argv[1:]:
p = Path(path_str).resolve()
file_type = detect_file_type(p)
compress = should_compress(p)
print(f" {p.name:30s} type={file_type:20s} compress={compress}")
scripts/validate.py
#!/usr/bin/env python3
import re
from collections import Counter
from pathlib import Path
URL_REGEX = re.compile(r"https?://[^\s)`]+") # a backtick is never part of a URL
FENCE_OPEN_REGEX = re.compile(r"^(\s{0,3})(`{3,}|~{3,})(.*)$")
# A line that is nothing but a fence marker plus an optional info string, at ANY
# indentation. Used ONLY to scrub leaked markers before inline-code pairing (see
# extract_inline_codes) — never for block extraction.
#
# Widening FENCE_OPEN_REGEX itself to `\s*` looks like the obvious fix for #820
# and is a net regression: a lone indented ``` (the natural way to SHOW a fence
# inside prose) then opens a block that runs to EOF, swallowing real code blocks
# and silently removing their inline spans from validation. That turns a
# false-failure bug into a false-PASS bug, and a false PASS overwrites the
# user's file with unvalidated output.
FENCE_MARKER_LINE_REGEX = re.compile(r"^\s*(?:`{3,}|~{3,})[^`~]*$")
# Cap on how much of a lost/added span is echoed in an error message. Unpaired
# backticks can make a "span" hundreds of characters of prose; printing it whole
# is what made #820's failures undiagnosable.
MAX_REPORTED_SPAN = 60
HEADING_REGEX = re.compile(r"^(#{1,6})\s+(.*)", re.MULTILINE)
BULLET_REGEX = re.compile(r"^\s*[-*+]\s+", re.MULTILINE)
# Any list item, ordered or not. Four spaces inside a list item is the item's
# content indentation, never an indented code block.
LIST_ITEM_REGEX = re.compile(r"^\s*(?:[-*+]|\d+[.)])\s")
# crude but effective path detection
# Requires either a path prefix (./ ../ / or drive letter) or a slash/backslash within the match
PATH_REGEX = re.compile(r"(?:\./|\.\./|/|[A-Za-z]:\\)[\w\-/\\\.]+|[\w\-\.]+[/\\][\w\-/\\\.]+")
# PATH_REGEX is crude on purpose and also matches ordinary prose pairs —
# "pros/cons", "Node/browser", "state/lifecycle". Caveman prose ADDS those
# constructions freely and dropping one breaks nothing, so only an unambiguous
# path — a leading ./ ../ / or drive letter, or a dotted filename in the last
# component — is treated as a hard loss.
DEFINITE_PATH_REGEX = re.compile(r"^(?:\./|\.\./|/|[A-Za-z]:\\)|[^/\\]*\.[A-Za-z0-9]{1,8}$")
class ValidationResult:
def __init__(self):
self.is_valid = True
self.errors = []
self.warnings = []
def add_error(self, msg):
self.is_valid = False
self.errors.append(msg)
def add_warning(self, msg):
self.warnings.append(msg)
def read_file(path: Path) -> str:
return path.read_text(encoding="utf-8")
# ---------- Extractors ----------
def extract_headings(text):
return [(level, title.strip()) for level, title in HEADING_REGEX.findall(text)]
def extract_code_blocks(text):
"""Line-based fenced code block extractor.
Handles ``` and ~~~ fences with variable length (CommonMark: closing
fence must use same char and be at least as long as opening). Supports
nested fences (e.g. an outer 4-backtick block wrapping inner 3-backtick
content).
Merges fenced and indented blocks by their DOCUMENT POSITION, not by
type. validate_code_blocks compares this list positionally (c1 != c2),
so a type-grouped return (all fenced, then all indented) makes two
blocks that swapped RELATIVE order in the document produce the same
list on both sides: the comparison passes while the invariant it
exists to check (block order preserved) has actually been violated.
"""
blocks = []
lines = text.split("\n")
i = 0
n = len(lines)
while i < n:
m = FENCE_OPEN_REGEX.match(lines[i])
if not m:
i += 1
continue
start = i
fence_char = m.group(2)[0]
fence_len = len(m.group(2))
open_line = lines[i]
block_lines = [open_line]
i += 1
closed = False
while i < n:
close_m = FENCE_OPEN_REGEX.match(lines[i])
if (
close_m
and close_m.group(2)[0] == fence_char
and len(close_m.group(2)) >= fence_len
and close_m.group(3).strip() == ""
):
block_lines.append(lines[i])
closed = True
i += 1
break
block_lines.append(lines[i])
i += 1
if closed:
blocks.append((start, "\n".join(block_lines)))
# Unclosed fences are silently skipped — they indicate malformed markdown
# and including them would cause false-positive validation failures.
ordered = sorted(blocks + extract_indented_code_blocks(text), key=lambda pair: pair[0])
return [block_text for _, block_text in ordered]
def extract_indented_code_blocks(text):
"""CommonMark indented code blocks — 4-space-indented runs outside any fence.
Without these, ` kubectl delete pod --all -n prod` was prose to the
validator: "code blocks preserved exactly" compared empty to empty and
PASSED while the compressor rewrote the command to
`kubectl delete pod -n dev`. A clean pass on a mutated destructive command
is the worst failure this tool has, because it overwrites the user's file.
Deliberately conservative about lists: inside a list item, four spaces are
the item's content indentation, not code, and nested bullets are ordinary
prose the compressor SHOULD rewrite. A run is only treated as code when the
document is not inside a list and the run is preceded by a blank line — so
this adds detections, it never turns existing passes into false failures.
Returns (start_line, text) pairs rather than bare text: extract_code_blocks
needs each block's document position to merge it with fenced blocks in
order (see that function's docstring).
"""
blocks = []
lines = text.split("\n")
fenced = set()
for block in extract_fenced_spans(lines):
fenced.update(block)
in_list = False
previous_blank = True
i = 0
n = len(lines)
while i < n:
line = lines[i]
stripped = line.strip()
if i in fenced:
in_list, previous_blank = in_list, False
i += 1
continue
if not stripped:
previous_blank = True
i += 1
continue
indent = len(line) - len(line.lstrip(" \t"))
if LIST_ITEM_REGEX.match(line):
in_list = True
elif indent == 0:
in_list = False
if not in_list and previous_blank and indent >= 4:
start = i
run = []
while i < n and i not in fenced:
current = lines[i]
if not current.strip():
# A blank line continues an indented block only if more
# indented content follows.
lookahead = i + 1
while lookahead < n and not lines[lookahead].strip():
lookahead += 1
if lookahead < n and lookahead not in fenced and \
len(lines[lookahead]) - len(lines[lookahead].lstrip(" \t")) >= 4:
run.extend(lines[i:lookahead])
i = lookahead
continue
break
if len(current) - len(current.lstrip(" \t")) < 4:
break
run.append(current)
i += 1
if run:
blocks.append((start, "\n".join(run)))
previous_blank = False
continue
previous_blank = False
i += 1
return blocks
def extract_fenced_spans(lines):
"""Line-index ranges covered by fenced blocks, so indented-code detection
never reaches inside one."""
spans = []
i = 0
n = len(lines)
while i < n:
m = FENCE_OPEN_REGEX.match(lines[i])
if not m:
i += 1
continue
fence_char = m.group(2)[0]
fence_len = len(m.group(2))
start = i
i += 1
while i < n:
close_m = FENCE_OPEN_REGEX.match(lines[i])
if (
close_m
and close_m.group(2)[0] == fence_char
and len(close_m.group(2)) >= fence_len
and close_m.group(3).strip() == ""
):
i += 1
break
i += 1
spans.append(range(start, i))
return spans
# Sentence punctuation right after a URL or path ("see docs/a.md.") is matched
# by both regexes but is not part of it; compression often moves it.
TRAILING_PUNCTUATION = ".,;:!?"
def extract_urls(text):
return {url.rstrip(TRAILING_PUNCTUATION) for url in URL_REGEX.findall(text)}
def extract_paths(text):
return {path.rstrip(TRAILING_PUNCTUATION) for path in PATH_REGEX.findall(text)}
def count_bullets(text):
return len(BULLET_REGEX.findall(text))
def extract_inline_codes(text):
"""Backtick-delimited inline spans, with fenced code blocks stripped first.
Previously used a column-0-anchored regex to strip fences, which misses
fences indented 1-3 spaces (valid CommonMark). Reuse extract_code_blocks
(FENCE_OPEN_REGEX-based, indentation-aware) instead so an indented fence's
body backticks don't leak into inline-code pairing.
Any fence-marker line that survives that pass is then blanked (#820). A
fence indented 4+ spaces — what you get from showing an example inside a
bullet — is not matched by FENCE_OPEN_REGEX, so extract_code_blocks does
not remove it and its OWN backticks used to leak in and shift the pairing
of every following span, making the file permanently uncompressible.
Blanking just the marker lines fixes that without removing any prose, and
cannot run away the way a widened fence opener does.
The span pattern deliberately still spans newlines. CommonMark permits a
line ending inside a code span and hard-wrapped markdown produces them, so
a single-line pattern silently drops real spans — which downgrades a
deleted or mutated span from error to PASS. Long/garbled spans are a
presentation problem, handled by truncating in the error message instead.
"""
text_without_fences = text
for block in extract_code_blocks(text):
text_without_fences = text_without_fences.replace(block, "", 1)
text_without_fences = "\n".join(
"" if FENCE_MARKER_LINE_REGEX.match(line) else line
for line in text_without_fences.split("\n")
)
# Run-aware pairing (CommonMark): a span opens with a backtick run and
# closes with an equal-length run. A mid-line ``` in prose ("inside a
# ```diff fence") has no equal closer, so it matches nothing — instead of
# its trailing backtick shifting the pairing of every following span.
# The opener is the WHOLE run (a lookahead refuses one more backtick), so a lone ``` can never
# shrink to a single backtick and pair with the next span. The content may
# hold a run of a different length (``a`b``), so it is "anything, lazily"
# up to the first equal run, not "anything but a backtick"; DOTALL keeps
# hard-wrapped spans in the comparison set.
return [
m.group(2)
# \x60 is a backtick: spelled out so no "!" sits before a literal one.
for m in re.finditer(r"(?<!\x60)(\x60+)(?!\x60)(.+?)(?<!\x60)\1(?!\x60)", text_without_fences, re.DOTALL)
]
# ---------- Validators ----------
def validate_headings(orig, comp, result):
h1 = extract_headings(orig)
h2 = extract_headings(comp)
# Changed heading TEXT is an error, not a warning. Every in-document anchor
# link points at a heading's slug, so renaming "# Configuration Options" to
# "# Config" silently breaks all of them — and SKILL.md and CLAUDE.md both
# state headings are preserved. Only counts used to gate the overwrite, so
# a run that renamed every heading reported "Validation passed". A level-only
# change keeps every slug intact and stays a warning.
if len(h1) != len(h2):
result.add_error(f"Heading count mismatch: {len(h1)} vs {len(h2)}")
return
t1 = [text for _, text in h1]
t2 = [text for _, text in h2]
if t1 != t2:
lost = [t for t in t1 if t not in t2]
added = [t for t in t2 if t not in t1]
result.add_error(f"Heading text/order changed: lost={lost}, added={added}")
elif h1 != h2:
# Same text, different level. The outline moved but no anchor broke —
# slugs come from the text, so links still resolve.
result.add_warning("Heading levels changed")
def validate_code_blocks(orig, comp, result):
c1 = extract_code_blocks(orig)
c2 = extract_code_blocks(comp)
if c1 != c2:
result.add_error("Code blocks not preserved exactly")
def validate_urls(orig, comp, result):
u1 = extract_urls(orig)
u2 = extract_urls(comp)
if u1 != u2:
result.add_error(f"URL mismatch: lost={u1 - u2}, added={u2 - u1}")
def validate_paths(orig, comp, result):
"""File paths are preserved — an error, never a warning.
This validator never called add_error at all, so a compressed file that had
dropped `src/hooks/caveman-config.js` still reported "Validation passed" and
the in-place overwrite stood. SKILL.md and CLAUDE.md both promise paths
survive compression; nothing enforced it.
"""
p1 = extract_paths(orig)
p2 = extract_paths(comp)
lost = p1 - p2
added = p2 - p1
definite = {p for p in lost if DEFINITE_PATH_REGEX.search(p)}
if definite:
result.add_error(f"File paths lost: {sorted(definite)}")
if (lost - definite) or added:
result.add_warning(f"Path mismatch: lost={sorted(lost)}, added={sorted(added)}")
def validate_bullets(orig, comp, result):
b1 = count_bullets(orig)
b2 = count_bullets(comp)
if b1 == 0:
return
diff = abs(b1 - b2) / b1
if diff > 0.15:
result.add_warning(f"Bullet count changed too much: {b1} -> {b2}")
def validate_inline_codes(orig, comp, result):
def _render_spans(spans):
"""Render spans for an error message, truncated and newline-escaped.
A span may legitimately contain newlines, and an unpaired backtick can
make one hundreds of characters of prose. Printing those whole is what
made #820's failures undiagnosable — but the fix belongs here, in
presentation, not in what counts as a span.
"""
out = []
for span in sorted(spans):
flat = span.replace("\n", "\\n")
if len(flat) > MAX_REPORTED_SPAN:
flat = flat[:MAX_REPORTED_SPAN] + "…"
out.append(repr(flat))
return "{" + ", ".join(out) + "}"
c1 = Counter(extract_inline_codes(orig))
c2 = Counter(extract_inline_codes(comp))
if c1 != c2:
lost = set(c1.keys()) - set(c2.keys())
added = set(c2.keys()) - set(c1.keys())
for code, count in c1.items():
if code in c2 and c2[code] < count:
lost.add(f"{code} (lost {count - c2[code]} of {count} occurrences)")
if lost:
result.add_error(f"Inline code lost: {_render_spans(lost)}")
if added:
result.add_warning(f"Inline code added: {_render_spans(added)}")
# ---------- Main ----------
def validate(original_path: Path, compressed_path: Path) -> ValidationResult:
result = ValidationResult()
orig = read_file(original_path)
comp = read_file(compressed_path)
validate_headings(orig, comp, result)
validate_code_blocks(orig, comp, result)
validate_urls(orig, comp, result)
validate_paths(orig, comp, result)
validate_bullets(orig, comp, result)
validate_inline_codes(orig, comp, result)
return result
# ---------- CLI ----------
if __name__ == "__main__":
import sys
if len(sys.argv) != 3:
print("Usage: python validate.py <original> <compressed>")
sys.exit(1)
orig = Path(sys.argv[1]).resolve()
comp = Path(sys.argv[2]).resolve()
res = validate(orig, comp)
print(f"\nValid: {res.is_valid}")
if res.errors:
print("\nErrors:")
for e in res.errors:
print(f" - {e}")
if res.warnings:
print("\nWarnings:")
for w in res.warnings:
print(f" - {w}")
SECURITY.md
# Security
## Snyk High Risk Rating
`caveman-compress` receives a Snyk High Risk rating due to static analysis heuristics. This document explains what the skill does and does not do.
### What triggers the rating
1. **subprocess usage**: The skill calls the `claude` CLI via `subprocess.run()` as a fallback when `ANTHROPIC_API_KEY` is not set. It can also call `opencode run` when `CAVEMAN_COMPRESS_PROVIDER=opencode`. Subprocess calls use fixed argument lists — no shell interpolation occurs. Claude receives user file content via stdin; opencode receives the generated prompt through a temporary `--file` attachment that is deleted after the call.
The file being compressed is untrusted input, and opencode 2.x lets its agents use every tool that is not explicitly denied: shell, file edits, web search and fetch, MCP tools from your opencode config, subagents. Compression needs none of them. Compress therefore runs `opencode run --standalone --agent caveman-compress` with `OPENCODE_CONFIG_CONTENT` denying every permission (`"*": "deny"`) both globally and for that agent, and never passes `--auto`. The dedicated agent matters because opencode applies a per-agent rule after the global one, so an `agent.build.permission` allow in your own config would otherwise re-enable a tool. `--standalone` matters because the shared opencode background service does not read the caller's environment, so the deny rule only binds a private server. Checked on opencode 2.0.22: the model is offered no tools and the prompt attachment is still read. Unlike the Claude path (`--strict-mcp-config`), compress does not switch off the MCP servers in your opencode config: their tools are denied, but opencode may still start them.
2. **File read/write**: The skill reads the file the user explicitly points it at, compresses it, and writes the result back to the same path. A `.original.md` backup is saved to an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups/<parent-dir-name>-<hash>/`, or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>-<hash>\` on Windows, where `<hash>` keys the file's full parent path so two projects' same-named files never share a backup). The opencode provider writes a temporary prompt attachment in the OS temp directory and deletes it after the subprocess exits. Beyond the target file, that backup location, and the temporary attachment, no files are read or written.
### What the skill does NOT do
- Does not execute user file content as code
- Does not make network requests except through the configured LLM provider CLI, SDK, or (for `openai-compat`) the one configured HTTP endpoint
- Does not read unrelated files outside the path the user provides
- Does not use shell=True or string interpolation in subprocess calls
- Does not collect or transmit any data beyond the file being compressed
### Auth behavior
Default path uses Claude. If `ANTHROPIC_API_KEY` is set, the skill uses the Anthropic Python SDK directly (no subprocess). If not set, it falls back to the `claude` CLI, which uses the user's existing Claude desktop authentication.
Set `CAVEMAN_COMPRESS_PROVIDER=opencode` to use `opencode run` instead. Set `CAVEMAN_COMPRESS_MODEL=provider/model` for compress-specific model selection. `CAVEMAN_MODEL` is the fallback when `CAVEMAN_COMPRESS_MODEL` is unset.
Set `CAVEMAN_COMPRESS_PROVIDER=openai-compat` to POST the prompt with Python's standard-library `urllib` (no subprocess, no extra dependency) to `$CAVEMAN_COMPRESS_ENDPOINT/chat/completions`, default `http://localhost:11434/v1` (Ollama). Other local servers: LM Studio `http://localhost:1234/v1`, llama.cpp `http://localhost:8080/v1`, vLLM `http://localhost:8000/v1`. `CAVEMAN_COMPRESS_API_KEY`, if set, is sent as a Bearer token. A non-local `http://` endpoint sends the file and that key in plaintext; use `https://` for anything off the machine.
### File size limit
Files larger than 500KB are rejected before any API call is made.
### Snyk W007 ("insecure credential handling")
W007 fires on the rule that code blocks, commands, and exact error strings stay verbatim. See [Scanner warnings](../../SECURITY.md#scanner-warnings) in the repository SECURITY.md.
### Reporting a vulnerability
If you believe you've found a genuine security issue, please open a GitHub issue with the label `security`.
SKILL.md
---
name: caveman-compress
description: >
Compress a memory file such as CLAUDE.md or a todo list into caveman format
to save input tokens, keeping a readable backup. Sends the file's text to the
configured model provider. Trigger: /caveman-compress.
---
# Caveman Compress
## Purpose
> **Data leaves the machine.** Compression sends the file's full text to the model provider that `scripts/compress.py` is configured for: the `claude` CLI, the Anthropic API, or an OpenAI-compatible endpoint. Files whose names look like secrets (`.env`, `credentials`, `secrets`, `.netrc`, …) are refused before they are read, and a file whose text holds a credential (a private key block, or a cloud, Git host or model API token) is refused before anything is sent. Don't compress a file the provider shouldn't see.
Compress natural language files (CLAUDE.md, todos, preferences) into caveman-speak to reduce input tokens. Compressed version overwrites original. Human-readable backup saved as `<filename>.original.md`, but NOT beside the source file — it lives in an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups/<parent-dir-name>/`, or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>\` on Windows) so skill auto-loaders don't re-ingest it as a live file.
## Trigger
`/caveman-compress <filepath>` or when user asks to compress a memory file.
**Requires an explicit filepath.** If the user pastes raw text instead of naming a file, do NOT improvise an inline compression — ask them for the filepath to compress, or offer to save the pasted text to a file first. Writing your own reasoning/commentary about the request into the compressed output (e.g. "User passed text directly, not a filepath...") is the exact failure this guards against — never let that leak into a saved file.
## Process
1. The compression scripts live in `scripts/` (adjacent to this SKILL.md). If the path is not immediately available, search for `scripts/__main__.py` next to this SKILL.md.
2. From the directory containing this SKILL.md, run:
python3 -m scripts <absolute_filepath>
3. The CLI will:
- detect file type (no tokens)
- call Claude to compress
- validate output (no tokens)
- if errors: cherry-pick fix with Claude (targeted fixes only, no recompression)
- retry up to 2 times
- if still failing after 2 retries: report error to user, leave original file untouched
4. Return result to user
## Compression Rules
### Remove
- Articles: a, an, the
- Filler: just, really, basically, actually, simply, essentially, generally
- Pleasantries: "sure", "certainly", "of course", "happy to", "I'd recommend"
- Hedging: "it might be worth", "you could consider", "it would be good to"
- Redundant phrasing: "in order to" → "to", "make sure to" → "ensure", "the reason is because" → "because"
- Connective fluff: "however", "furthermore", "additionally", "in addition"
### Preserve EXACTLY (never modify)
- Code blocks (fenced ``` and indented)
- Inline code (`backtick content`)
- URLs and links (full URLs, markdown links)
- File paths (`/src/components/...`, `./config.yaml`)
- Commands (`npm install`, `git commit`, `docker build`)
- Technical terms (library names, API names, protocols, algorithms)
- Proper nouns (project names, people, companies)
- Dates, version numbers, numeric values
- Environment variables (`$HOME`, `NODE_ENV`)
### Preserve Structure
- All markdown headings (keep exact heading text, compress body below)
- Bullet point hierarchy (keep nesting level)
- Numbered lists (keep numbering)
- Tables (compress cell text, keep structure)
- Frontmatter/YAML headers in markdown files
### Compress
- Use short synonyms: "big" not "extensive", "fix" not "implement a solution for", "use" not "utilize"
- Fragments OK: "Run tests before commit" not "You should always run tests before committing"
- Drop "you should", "make sure to", "remember to" — just state the action
- Merge redundant bullets that say the same thing differently
- Keep one example where multiple examples show the same pattern
CRITICAL RULE:
Anything inside ``` ... ``` must be copied EXACTLY.
Do not:
- remove comments
- remove spacing
- reorder lines
- shorten commands
- simplify anything
Inline code (`...`) must be preserved EXACTLY.
Do not modify anything inside backticks.
If file contains code blocks:
- Treat code blocks as read-only regions
- Only compress text outside them
- Do not merge sections around code
## Pattern
Original:
> You should always make sure to run the test suite before pushing any changes to the main branch. This is important because it helps catch bugs early and prevents broken builds from being deployed to production.
Compressed:
> Run tests before push to main. Catch bugs early, prevent broken prod deploys.
Original:
> The application uses a microservices architecture with the following components. The API gateway handles all incoming requests and routes them to the appropriate service. The authentication service is responsible for managing user sessions and JWT tokens.
Compressed:
> Microservices architecture. API gateway route all requests to services. Auth service manage user sessions + JWT tokens.
## Boundaries
- ONLY compress natural language files (.md, .txt, .typ, .typst, .tex, extensionless)
- NEVER modify: .py, .js, .ts, .json, .yaml, .yml, .toml, .env, .lock, .css, .html, .xml, .sql, .sh
- If file has mixed content (prose + code), compress ONLY the prose sections
- If unsure whether something is code or prose, leave it unchanged
- Original file is backed up as FILE.original.md before overwriting — in the out-of-tree backup data dir (see Purpose), not beside the source file
- Never compress FILE.original.md (skip it)