612 lines
20 KiB
Python
612 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""CASAN-native deterministic context compressor (Plan-08 Track 3).
|
|
|
|
The compressor reduces long tool output before it enters model context while
|
|
preserving operationally significant lines. It is deliberately not a tokenizer
|
|
or a billing source: token figures are whitespace-based estimates and are
|
|
labelled as such.
|
|
|
|
Safety contract:
|
|
* preserve severity-tagged errors/warnings, failure/skip signals, summaries,
|
|
operational result lines, and project-supplied must-keep patterns;
|
|
* verify preservation after compression, including match multiplicity;
|
|
* on a preservation failure, return non-zero and emit raw input by default so
|
|
a caller that ignores the exit code still cannot consume a lossy view;
|
|
* support halt-with-no-output for callers whose policy forbids raw fallback;
|
|
* optionally emit a hash-bound JSON evidence report;
|
|
* never compress a failed command when ``--failed`` is supplied.
|
|
|
|
Governance ordering remains the caller's responsibility:
|
|
H4 scan raw -> H5 hash raw -> compress -> H3 preservation/faithfulness
|
|
-> H4 scan compressed -> H5 bind raw/compressed hashes -> model.
|
|
"""
|
|
|
|
import argparse
|
|
from collections import Counter
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
from typing import Dict, Iterable, List, Pattern, Sequence, Tuple
|
|
|
|
|
|
DEFAULT_MAX_BYTES = 2 * 1024 * 1024
|
|
PatternRule = Tuple[str, Pattern[str]]
|
|
|
|
|
|
def _casan_app_root() -> str:
|
|
"""Find the project/source root without relying on a fixed file depth."""
|
|
directory = os.path.abspath(os.path.dirname(__file__))
|
|
current = directory
|
|
while current != os.path.dirname(current):
|
|
if (
|
|
os.path.isdir(os.path.join(current, ".specify"))
|
|
or os.path.isdir(os.path.join(current, "packages/casan-harness"))
|
|
):
|
|
return current
|
|
current = os.path.dirname(current)
|
|
return os.path.abspath(os.path.join(directory, "..", "..", ".."))
|
|
|
|
|
|
def compression_enabled() -> bool:
|
|
"""Return the governed ``compression.enabled`` setting.
|
|
|
|
An absent or invalid store retains the historical enabled-by-default
|
|
behavior. A malformed project-supplied must-keep policy is handled
|
|
separately and fails closed.
|
|
"""
|
|
store_file = os.environ.get(
|
|
"CASAN_CP_STORE_FILE",
|
|
os.path.join(
|
|
_casan_app_root(),
|
|
".specify/level5/control-plane-settings.json",
|
|
),
|
|
)
|
|
if not os.path.isfile(store_file):
|
|
return True
|
|
try:
|
|
with open(store_file, encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
setting = data.get("settings", {}).get("compression.enabled")
|
|
return True if setting is None else bool(setting["value"])
|
|
except (OSError, ValueError, KeyError, TypeError):
|
|
return True
|
|
|
|
|
|
# Serilog/log4net-style severity tags need explicit recognition. Word-boundary
|
|
# matching alone does not classify abbreviations such as ``[WRN]``.
|
|
SEVERITY_TAG_RE = re.compile(
|
|
r"\[(ERR(?:OR)?|FTL|FATAL|CRIT(?:ICAL)?|WRN|WARN(?:ING)?)\]",
|
|
re.IGNORECASE,
|
|
)
|
|
IMPORTANT_RE = re.compile(
|
|
r"\b("
|
|
r"error|errors|fail|failed|failure|failing|exception|panic|"
|
|
r"denied|blocked|warn|warning|timeout|timed out|"
|
|
r"abort|aborted|cancel|cancelled|skip|skipped|incomplete|"
|
|
r"partially completed|partial (?:result|output|analysis|run|failure|success)"
|
|
r")\b",
|
|
re.IGNORECASE,
|
|
)
|
|
SUMMARY_RE = re.compile(
|
|
r"\b(\d+)\s+(pass|passed|fail|failed|tests?|errors?|warnings?|files?|artifacts?)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
OPERATIONAL_RE = re.compile(
|
|
r"\b("
|
|
r"exit\s*code|return\s*code|"
|
|
r"output\s*(?:path|directory)|"
|
|
r"generated\s+\d+\s+(?:files?|artifacts?)"
|
|
r")\b",
|
|
re.IGNORECASE,
|
|
)
|
|
DEDUP_SUFFIX_RE = re.compile(r"^(.*) \(x([1-9][0-9]*)\)$")
|
|
|
|
|
|
def configured_max_bytes() -> int:
|
|
raw = os.environ.get("CASAN_MAX_INPUT_BYTES", str(DEFAULT_MAX_BYTES))
|
|
try:
|
|
value = int(raw)
|
|
except ValueError:
|
|
print(
|
|
"COMPRESS_FAIL invalid_CASAN_MAX_INPUT_BYTES fail-closed",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(1)
|
|
if value <= 0:
|
|
print(
|
|
"COMPRESS_FAIL non_positive_CASAN_MAX_INPUT_BYTES fail-closed",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(1)
|
|
return value
|
|
|
|
|
|
def read_capped(src: str, max_bytes: int) -> Tuple[str, int]:
|
|
"""Read at most ``max_bytes`` and return text plus replacement count."""
|
|
try:
|
|
if src == "-":
|
|
data = sys.stdin.buffer.read(max_bytes + 1)
|
|
else:
|
|
with open(src, "rb") as handle:
|
|
data = handle.read(max_bytes + 1)
|
|
except OSError as exc:
|
|
print(f"COMPRESS_FAIL unreadable_input: {exc}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
if len(data) > max_bytes:
|
|
print(
|
|
f"COMPRESS_FAIL input_exceeds_cap({max_bytes}B) fail-closed",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(1)
|
|
text = data.decode("utf-8", errors="replace")
|
|
return text, text.count("\ufffd")
|
|
|
|
|
|
def load_patterns(path: str) -> List[PatternRule]:
|
|
"""Load and compile one regex per line, ignoring blank/comment lines."""
|
|
if not path:
|
|
return []
|
|
try:
|
|
with open(path, encoding="utf-8", errors="replace") as handle:
|
|
raw_rules = [
|
|
(line_number, line.strip())
|
|
for line_number, line in enumerate(handle, start=1)
|
|
if line.strip() and not line.lstrip().startswith("#")
|
|
]
|
|
except OSError as exc:
|
|
print(f"COMPRESS_FAIL must_keep_file_unreadable: {exc}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
rules: List[PatternRule] = []
|
|
for line_number, expression in raw_rules:
|
|
try:
|
|
rules.append((expression, re.compile(expression)))
|
|
except re.error as exc:
|
|
print(
|
|
"COMPRESS_FAIL invalid_must_keep_regex "
|
|
f"file={path} line={line_number}: {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(1)
|
|
return rules
|
|
|
|
|
|
def is_must_keep(line: str, patterns: Sequence[PatternRule]) -> bool:
|
|
return any(pattern.search(line) for _, pattern in patterns)
|
|
|
|
|
|
def is_protected(line: str, patterns: Sequence[PatternRule]) -> bool:
|
|
"""Return whether a line is forbidden from disappearing."""
|
|
return bool(
|
|
SEVERITY_TAG_RE.search(line)
|
|
or IMPORTANT_RE.search(line)
|
|
or SUMMARY_RE.search(line)
|
|
or OPERATIONAL_RE.search(line)
|
|
or is_must_keep(line, patterns)
|
|
)
|
|
|
|
|
|
def estimate_tokens(text: str) -> int:
|
|
"""Return a whitespace estimate, never provider/billed token usage."""
|
|
return len(text.split())
|
|
|
|
|
|
def dedup(lines: Sequence[str]) -> List[str]:
|
|
"""Collapse consecutive duplicate lines while retaining multiplicity."""
|
|
output: List[str] = []
|
|
index = 0
|
|
count_lines = len(lines)
|
|
while index < count_lines:
|
|
end = index
|
|
while end + 1 < count_lines and lines[end + 1] == lines[index]:
|
|
end += 1
|
|
occurrences = end - index + 1
|
|
output.append(
|
|
lines[index]
|
|
if occurrences == 1
|
|
else f"{lines[index]} (x{occurrences})"
|
|
)
|
|
index = end + 1
|
|
return output
|
|
|
|
|
|
def compress(
|
|
text: str,
|
|
mode: str,
|
|
must_keep: Sequence[PatternRule],
|
|
) -> List[str]:
|
|
"""Create a candidate compressed view.
|
|
|
|
Preservation is verified independently after this function returns.
|
|
"""
|
|
lines = compression_source_lines(text.split("\n"), mode, must_keep)
|
|
if mode in {"dedup", "structural"}:
|
|
return dedup(lines)
|
|
if mode == "extractive":
|
|
return lines
|
|
raise ValueError(f"unknown mode: {mode}")
|
|
|
|
|
|
def compression_source_lines(
|
|
raw_lines: Sequence[str],
|
|
mode: str,
|
|
must_keep: Sequence[PatternRule],
|
|
) -> List[str]:
|
|
"""Return the exact raw lines from which a candidate may be built."""
|
|
if mode == "dedup":
|
|
return list(raw_lines)
|
|
return [line for line in raw_lines if is_protected(line, must_keep)]
|
|
|
|
|
|
def expanded_line_counts(
|
|
lines: Iterable[str],
|
|
dedup_encoded: bool,
|
|
source_lines: Sequence[str],
|
|
) -> Counter:
|
|
"""Decode ``(xN)`` markers produced by ``dedup`` into weighted counts."""
|
|
candidate_lines = list(lines)
|
|
# This is the normal path and resolves the otherwise ambiguous case where
|
|
# a real log line itself ends in ``(xN)``.
|
|
if dedup_encoded and candidate_lines == dedup(source_lines):
|
|
return Counter(source_lines)
|
|
|
|
counts: Counter = Counter()
|
|
for line in candidate_lines:
|
|
match = DEDUP_SUFFIX_RE.match(line) if dedup_encoded else None
|
|
if match:
|
|
counts[match.group(1)] += int(match.group(2))
|
|
else:
|
|
counts[line] += 1
|
|
return counts
|
|
|
|
|
|
def protected_line_deficits(
|
|
source_lines: Sequence[str],
|
|
candidate_lines: Sequence[str],
|
|
dedup_encoded: bool,
|
|
) -> Dict[str, int]:
|
|
"""Return exact protected-line multiplicity missing from the candidate."""
|
|
expected = Counter(source_lines)
|
|
actual = expanded_line_counts(
|
|
candidate_lines,
|
|
dedup_encoded=dedup_encoded,
|
|
source_lines=source_lines,
|
|
)
|
|
return {
|
|
line: expected_count - actual.get(line, 0)
|
|
for line, expected_count in expected.items()
|
|
if actual.get(line, 0) < expected_count
|
|
}
|
|
|
|
|
|
def required_pattern_deficits(
|
|
raw_lines: Sequence[str],
|
|
source_lines: Sequence[str],
|
|
candidate_lines: Sequence[str],
|
|
required: Sequence[PatternRule],
|
|
dedup_encoded: bool,
|
|
) -> List[Dict[str, object]]:
|
|
"""Verify every required match present in raw remains in the candidate.
|
|
|
|
A pattern absent from raw is not a failure: the invariant is preservation,
|
|
not fabrication. Match multiplicity prevents one surviving warning from
|
|
hiding the loss of a second warning matched by the same rule.
|
|
"""
|
|
deficits: List[Dict[str, object]] = []
|
|
actual_lines = expanded_line_counts(
|
|
candidate_lines,
|
|
dedup_encoded=dedup_encoded,
|
|
source_lines=source_lines,
|
|
)
|
|
for expression, pattern in required:
|
|
expected = sum(1 for line in raw_lines if pattern.search(line))
|
|
actual = sum(
|
|
count
|
|
for line, count in actual_lines.items()
|
|
if pattern.search(line)
|
|
)
|
|
if actual < expected:
|
|
deficits.append(
|
|
{
|
|
"pattern": expression,
|
|
"expected": expected,
|
|
"actual": actual,
|
|
"missing": expected - actual,
|
|
}
|
|
)
|
|
return deficits
|
|
|
|
|
|
def severity_counts(text: str) -> Dict[str, int]:
|
|
counts = {"error": 0, "warning": 0, "critical": 0}
|
|
for line in text.splitlines():
|
|
match = SEVERITY_TAG_RE.search(line)
|
|
if not match:
|
|
continue
|
|
level = match.group(1).upper()
|
|
if level.startswith(("ERR",)):
|
|
counts["error"] += 1
|
|
elif level.startswith(("WRN", "WARN")):
|
|
counts["warning"] += 1
|
|
else:
|
|
counts["critical"] += 1
|
|
return counts
|
|
|
|
|
|
def sha256_text(text: str) -> str:
|
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def deficit_fingerprints(deficits: Dict[str, int]) -> List[Dict[str, object]]:
|
|
"""Report hashes, not potentially sensitive raw lines."""
|
|
return [
|
|
{"line_sha256": sha256_text(line), "missing": missing}
|
|
for line, missing in sorted(deficits.items())
|
|
]
|
|
|
|
|
|
def write_json_report(path: str, report: Dict[str, object]) -> None:
|
|
"""Atomically write a private evidence report."""
|
|
if path == "-":
|
|
print(
|
|
"COMPRESS_FAIL report_json_stdout_conflicts_with_compressed_output",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(1)
|
|
absolute = os.path.abspath(path)
|
|
directory = os.path.dirname(absolute)
|
|
try:
|
|
os.makedirs(directory, exist_ok=True)
|
|
descriptor, temporary_path = tempfile.mkstemp(
|
|
prefix=".context-compress-",
|
|
suffix=".json.tmp",
|
|
dir=directory,
|
|
)
|
|
try:
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
|
json.dump(
|
|
report,
|
|
handle,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
handle.write("\n")
|
|
os.chmod(temporary_path, 0o600)
|
|
os.replace(temporary_path, absolute)
|
|
except Exception:
|
|
try:
|
|
os.unlink(temporary_path)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
except OSError as exc:
|
|
print(f"COMPRESS_FAIL report_write_failed: {exc}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--mode",
|
|
choices=["dedup", "extractive", "structural"],
|
|
default="structural",
|
|
)
|
|
parser.add_argument("--input", default="-", help="input file or - for stdin")
|
|
parser.add_argument(
|
|
"--must-keep-file",
|
|
default="",
|
|
help="file with one must-keep regex per line",
|
|
)
|
|
parser.add_argument(
|
|
"--failed",
|
|
action="store_true",
|
|
help="raw passthrough when the producing command failed",
|
|
)
|
|
parser.add_argument(
|
|
"--respect-policy",
|
|
action="store_true",
|
|
help="honor control-plane compression.enabled",
|
|
)
|
|
parser.add_argument(
|
|
"--require-must-keep-file",
|
|
default="",
|
|
help=(
|
|
"verify every match present in raw remains in the candidate; "
|
|
"match multiplicity is enforced"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--on-preservation-failure",
|
|
choices=["raw", "halt"],
|
|
default="raw",
|
|
help=(
|
|
"raw: emit raw input and return 1; "
|
|
"halt: emit nothing and return 1"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--report-json",
|
|
default="",
|
|
help="optional atomic JSON evidence report path",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
raw, decode_replacements = read_capped(
|
|
args.input,
|
|
configured_max_bytes(),
|
|
)
|
|
must_keep = load_patterns(args.must_keep_file)
|
|
required = load_patterns(args.require_must_keep_file)
|
|
raw_lines = raw.split("\n")
|
|
|
|
exit_code = 0
|
|
preservation_status = "not_applicable"
|
|
pattern_deficits: List[Dict[str, object]] = []
|
|
line_deficits: Dict[str, int] = {}
|
|
|
|
if args.failed:
|
|
candidate_text = raw
|
|
output_text = raw
|
|
mode_used = "passthrough"
|
|
decision = "failed-command-raw-passthrough"
|
|
saving_status = "not_compressed"
|
|
elif args.respect_policy and not compression_enabled():
|
|
candidate_text = raw
|
|
output_text = raw
|
|
mode_used = "policy-disabled"
|
|
decision = "policy-disabled-raw-passthrough"
|
|
saving_status = "not_compressed"
|
|
else:
|
|
source_lines = compression_source_lines(
|
|
raw_lines,
|
|
args.mode,
|
|
must_keep,
|
|
)
|
|
candidate_lines = compress(raw, args.mode, must_keep)
|
|
candidate_text = "\n".join(candidate_lines)
|
|
line_deficits = protected_line_deficits(
|
|
source_lines,
|
|
candidate_lines,
|
|
dedup_encoded=args.mode in {"dedup", "structural"},
|
|
)
|
|
pattern_deficits = required_pattern_deficits(
|
|
raw_lines,
|
|
source_lines,
|
|
candidate_lines,
|
|
required,
|
|
dedup_encoded=args.mode in {"dedup", "structural"},
|
|
)
|
|
empty_loss = bool(raw.strip()) and not candidate_text.strip()
|
|
preservation_failed = bool(
|
|
line_deficits or pattern_deficits or empty_loss
|
|
)
|
|
|
|
if preservation_failed:
|
|
preservation_status = "failed"
|
|
exit_code = 1
|
|
mode_used = (
|
|
"fallback-raw"
|
|
if args.on_preservation_failure == "raw"
|
|
else "halt"
|
|
)
|
|
decision = "compression-rejected"
|
|
saving_status = "rejected"
|
|
output_text = (
|
|
raw if args.on_preservation_failure == "raw" else ""
|
|
)
|
|
else:
|
|
preservation_status = "passed"
|
|
mode_used = args.mode
|
|
decision = "compressed"
|
|
saving_status = "estimate_only_quality_gate_required"
|
|
output_text = candidate_text
|
|
|
|
input_estimate = estimate_tokens(raw)
|
|
candidate_estimate = estimate_tokens(candidate_text)
|
|
output_estimate = estimate_tokens(output_text)
|
|
if saving_status == "rejected":
|
|
# A rejected candidate has no claimable saving even when halt policy
|
|
# intentionally emits zero bytes.
|
|
estimated_saved = 0
|
|
estimated_ratio = 1.0
|
|
else:
|
|
estimated_saved = input_estimate - output_estimate
|
|
estimated_ratio = (
|
|
round(output_estimate / input_estimate, 4)
|
|
if input_estimate
|
|
else 1.0
|
|
)
|
|
emitted_text = output_text
|
|
if emitted_text and not emitted_text.endswith("\n"):
|
|
emitted_text += "\n"
|
|
|
|
report: Dict[str, object] = {
|
|
"schema_version": 1,
|
|
"measurement_source": "whitespace_estimate",
|
|
"provider_telemetry": None,
|
|
"mode_requested": args.mode,
|
|
"mode_used": mode_used,
|
|
"decision": decision,
|
|
"exit_code": exit_code,
|
|
"saving_status": saving_status,
|
|
"preservation": {
|
|
"status": preservation_status,
|
|
"protected_line_missing": sum(line_deficits.values()),
|
|
"protected_line_deficits": deficit_fingerprints(line_deficits),
|
|
"required_pattern_deficits": pattern_deficits,
|
|
},
|
|
"estimated_tokens": {
|
|
"input": input_estimate,
|
|
"candidate": candidate_estimate,
|
|
"output": output_estimate,
|
|
"saved": estimated_saved,
|
|
"ratio": estimated_ratio,
|
|
},
|
|
"severity": {
|
|
"raw": severity_counts(raw),
|
|
"candidate": severity_counts(candidate_text),
|
|
"output": severity_counts(output_text),
|
|
},
|
|
"hashes": {
|
|
"raw_sha256": sha256_text(raw),
|
|
"candidate_sha256": sha256_text(candidate_text),
|
|
"output_sha256": sha256_text(emitted_text),
|
|
},
|
|
"bytes": {
|
|
"raw": len(raw.encode("utf-8")),
|
|
"candidate": len(candidate_text.encode("utf-8")),
|
|
"output": len(emitted_text.encode("utf-8")),
|
|
},
|
|
"decode_replacement_count": decode_replacements,
|
|
"requires_downstream_h3_quality_gate": decision == "compressed",
|
|
"requires_downstream_h4_compressed_scan": decision == "compressed",
|
|
"requires_downstream_h5_audit_binding": decision == "compressed",
|
|
}
|
|
|
|
if args.report_json:
|
|
write_json_report(args.report_json, report)
|
|
|
|
sys.stdout.write(emitted_text)
|
|
print(
|
|
"COMPRESS "
|
|
f"mode={mode_used} "
|
|
f"measurement_source=whitespace_estimate "
|
|
f"in_tokens={input_estimate} "
|
|
f"candidate_tokens={candidate_estimate} "
|
|
f"out_tokens={output_estimate} "
|
|
f"saved={estimated_saved} "
|
|
f"ratio={estimated_ratio} "
|
|
f"preservation={preservation_status} "
|
|
f"saving_status={saving_status} "
|
|
f"must_keep_missing={sum(int(item['missing']) for item in pattern_deficits)} "
|
|
f"protected_missing={sum(line_deficits.values())}",
|
|
file=sys.stderr,
|
|
)
|
|
if pattern_deficits:
|
|
details = ",".join(
|
|
f"{item['pattern']}({item['actual']}/{item['expected']})"
|
|
for item in pattern_deficits
|
|
)
|
|
print(f"COMPRESS_MUST_KEEP_DROPPED {details}", file=sys.stderr)
|
|
if line_deficits:
|
|
print(
|
|
"COMPRESS_PROTECTED_CONTENT_DROPPED "
|
|
f"occurrences={sum(line_deficits.values())}",
|
|
file=sys.stderr,
|
|
)
|
|
if preservation_status == "failed":
|
|
print(
|
|
"COMPRESS_REJECTED "
|
|
f"fallback={args.on_preservation_failure}",
|
|
file=sys.stderr,
|
|
)
|
|
return exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|