#!/usr/bin/env python3 """Classify stdin content and emit a sanitized, token-budget-friendly compression. The CLI never promises lossless *semantic* compression. It performs conservative, deterministic, content-type-aware shrinking (compact JSON, diff change-only views, log/search de-duplication, whitespace normalization) so large local output costs fewer tokens to keep in context. Secrets are redacted *before* the receipt is built, so no secret ever reaches the compressed body or the metadata. For exact byte-for-byte recovery the receipt points at `context-guard-artifact store`, which keeps the full sanitized content as a queryable local artifact. """ from __future__ import annotations import argparse import importlib.machinery import importlib.util import json import os from pathlib import Path import re import sys from typing import Callable, Iterable DEFAULT_MAX_BYTES = 10_000_000 MAX_MAX_BYTES = 100_000_000 MAX_SEARCH_DEDUPE_KEYS = 50_000 JSON_PARSE_FAILED = object() # 토큰 추정은 보수적 proxy 일 뿐이다(관측값 아님). 평균 ~4 chars/token 휴리스틱을 쓰되 # 메타데이터에 measurement="estimated" 로 명시해 관측 토큰 수와 혼동되지 않게 한다. TOKEN_PROXY_CHARS_PER_TOKEN = 4 CONTENT_TYPES = ("json", "diff", "log", "search", "code", "prose") COMPRESSION_MODES = ("conservative", "readable") READABLE_COMPRESSION_SCHEMA_VERSION = "contextguard.compress-readable.v1" READABLE_SENTENCE_LIMIT = 5 # diff 구조 라인(파일 헤더/헝크/변경)을 식별한다. 나머지 context 라인은 접어서 줄인다. DIFF_FILE_HEADER_RE = re.compile(r"^(diff --git |index [0-9a-f]|--- |\+\+\+ |rename |similarity |new file|deleted file)") DIFF_HUNK_RE = re.compile(r"^@@ .* @@") # search(grep/ripgrep) 라인: `path:line:content` 또는 `path:content`. # 콜론 앞 경로 토큰에 공백을 불허해, 타임스탬프 로그("2026-01-01 00:00:00 ...")가 # search 로 오분류되는 것을 막는다(로그 타임스탬프는 콜론 앞에 공백을 포함). SEARCH_LINE_RE = re.compile(r"^[^\s:][^:\n\s]*:(?:\d+:)?.") # log 시그널: 선두 타임스탬프나 로그 레벨 토큰. LOG_LEVEL_RE = re.compile(r"\b(TRACE|DEBUG|INFO|NOTICE|WARN|WARNING|ERROR|FATAL|CRITICAL)\b") LOG_TIMESTAMP_RE = re.compile(r"^\s*(?:\[)?\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}|^\s*\d{2}:\d{2}:\d{2}\b") # code 시그널: 흔한 소스 키워드/구두점. diff 와 겹치지 않도록 diff 판정을 먼저 한다. CODE_SIGNAL_RE = re.compile( r"(^\s*(def |class |function |func |import |from \S+ import |public |private |const |let |var |#include|package )" r"|[{};]\s*$|=>|::)" ) CODE_FENCE_RE = re.compile(r"(?m)^\s*```") JSON_KEY_RE = re.compile(r'"(?:[^"\\]|\\.)*"\s*:') QUOTED_STRING_RE = re.compile(r"""(?x) "(?:[^"\\]|\\.)*" | '(?:[^'\\]|\\.)*' """) HASH_RE = re.compile(r"\b(?:[0-9a-fA-F]{32,}|sha256:[0-9a-fA-F]{32,})\b") PATH_RE = re.compile( r"(?x)(?:" r"(?|]+\\)*[^\\\s:\"'<>|]+" r"|" r"\b[A-Za-z0-9._-]+\#path:[0-9a-f]{12}\b" r")" ) STACK_FRAME_RE = re.compile( r"(?m)^\s*(?:File\s+\"[^\"]+\",\s+line\s+\d+,\s+in\s+\S+|at\s+\S+.*\([^)]*:\d+(?::\d+)?\))" ) IDENTIFIER_RE = re.compile(r"\b[A-Za-z_][A-Za-z0-9_]*(?:[A-Z][A-Za-z0-9_]*)?\b") NUMERIC_CONSTANT_RE = re.compile(r"(? int: """Clamp an int-like value into [minimum, maximum], falling back on default.""" try: number = int(value) except (TypeError, ValueError, OverflowError): return default return min(max(number, minimum), maximum) class FallbackLineSanitizer: """Minimal secret scrubber used when the shared sanitizer cannot be loaded.""" SECRET_VALUE_RE = re.compile( r"(?i)(Bearer\s+\S+|Basic\s+\S+|gh[pousr]_[A-Za-z0-9_]{20,}|" r"github_pat_[A-Za-z0-9_]{20,}|xox[abprs]-[A-Za-z0-9-]{10,}|" r"sk-(?:ant|proj)-[A-Za-z0-9_-]{12,}|sk-[A-Za-z0-9][A-Za-z0-9_-]{20,}|" r"AIza[0-9A-Za-z_\-]{20,}|" r"([A-Za-z0-9_.-]*(?:api[_-]?key|token|secret|password|passwd|pwd)[A-Za-z0-9_.-]*\s*[:=]\s*)\S+)" ) def __init__( self, *, show_paths: bool = False, context: str = "unknown_text", ) -> None: self.show_paths = show_paths self.context = context self.redactions = 0 def sanitize(self, raw_line: str) -> tuple[str, bool]: def repl(match: re.Match[str]) -> str: groups = match.groups() if len(groups) >= 2 and groups[1]: return groups[1] + "[REDACTED]" return "[REDACTED]" line, count = self.SECRET_VALUE_RE.subn(repl, raw_line) if count: self.redactions += 1 return line, bool(count) def instantiate_line_sanitizer( factory: object, *, show_paths: bool, context: str, private_roots: tuple[str, ...] = (), ) -> object: try: return factory( # type: ignore[operator] show_paths=show_paths, context=context, private_roots=private_roots, ) except TypeError: if context != "unknown_text" or private_roots: raise RuntimeError( "adjacent sanitizer does not support required explicit context" ) return factory(show_paths=show_paths) # type: ignore[operator] def load_line_sanitizer( show_paths: bool, context: str = "unknown_text", private_roots: tuple[str, ...] = (), ) -> object: """Reuse the shipped strong sanitizer when present; else fall back locally. Mirrors context_escrow.py so the compress CLI redacts with the same rules as the rest of the kit when `sanitize_output.py` sits next to this script. """ script_dir = Path(__file__).resolve().parent for name in ("sanitize_output.py", "context-guard-sanitize-output", "claude-sanitize-output"): candidate = script_dir / name if not candidate.exists(): continue try: loader = importlib.machinery.SourceFileLoader(f"_context_guard_compress_sanitize_{os.getpid()}", str(candidate)) spec = importlib.util.spec_from_loader(loader.name, loader) if spec is None: raise RuntimeError("import spec unavailable") module = importlib.util.module_from_spec(spec) sys.modules[loader.name] = module try: loader.exec_module(module) except Exception: sys.modules.pop(loader.name, None) raise return instantiate_line_sanitizer( module.LineSanitizer, show_paths=show_paths, context=context, private_roots=private_roots, ) except Exception as exc: raise RuntimeError(f"could not load sanitizer {candidate}: {exc}") from exc return FallbackLineSanitizer(show_paths=show_paths, context=context) def sanitize_text( text: str, *, show_paths: bool = False, context: str = "unknown_text", private_roots: tuple[str, ...] = (), ) -> tuple[str, int]: """Redact secrets line-by-line, returning sanitized text and redacted-line count.""" sanitizer = load_line_sanitizer( show_paths, context=context, private_roots=private_roots, ) redacted = 0 out: list[str] = [] for line in text.splitlines(True): sanitized, did_redact = sanitizer.sanitize(line) # type: ignore[attr-defined] out.append(sanitized) if did_redact: redacted += 1 return "".join(out), redacted def read_bounded_stdin(max_bytes: int) -> tuple[str, bool, int]: """Read at most max_bytes from stdin, reporting truncation and bytes read.""" data = sys.stdin.buffer.read(max_bytes + 1) truncated = len(data) > max_bytes if truncated: data = data[:max_bytes] return data.decode("utf-8", errors="replace"), truncated, len(data) def line_count(text: str) -> int: """Count logical lines without an off-by-one on a trailing newline.""" if not text: return 0 return text.count("\n") + (0 if text.endswith("\n") else 1) def byte_length(text: str) -> int: """UTF-8 byte length using the same lossy decode policy as the rest of the kit.""" return len(text.encode("utf-8", errors="replace")) def token_proxy(text: str) -> int: """Conservative token estimate (chars/4). Labeled 'estimated' in metadata.""" if not text: return 0 return max(1, round(len(text) / TOKEN_PROXY_CHARS_PER_TOKEN)) LINE_BOUNDARY_CHARS = {"\n", "\r", "\v", "\f", "\x1c", "\x1d", "\x1e", "\x85", "\u2028", "\u2029"} def iter_text_lines(text: str) -> Iterable[str]: """Yield lines with str.splitlines() boundaries without building a line list.""" start = 0 index = 0 length = len(text) while index < length: char = text[index] if char == "\r" and index + 1 < length and text[index + 1] == "\n": yield text[start:index] index += 2 start = index continue if char in LINE_BOUNDARY_CHARS: yield text[start:index] index += 1 start = index continue index += 1 if start < length: yield text[start:] def sample_text_lines(text: str, limit: int) -> list[str]: sample: list[str] = [] for line in iter_text_lines(text): sample.append(line) if len(sample) >= limit: break return sample def classify_content(text: str) -> str: """Best-effort content classification into one of CONTENT_TYPES. Order matters: valid JSON and diff have the strongest unambiguous signals; search/log/code are sampled over the first lines; prose is the conservative default so unknown text is never over-compressed. """ stripped = text.strip() if not stripped: return "prose" if _looks_like_json(stripped): return "json" return classify_non_json_content(stripped) def classify_non_json_content(stripped: str) -> str: sample = sample_text_lines(stripped, 200) if _looks_like_diff(sample): return "diff" if _looks_like_search(sample): return "search" if _looks_like_log(sample): return "log" if _looks_like_code(sample): return "code" return "prose" def protected_zone_counts(text: str) -> dict[str, int]: """Conservatively count semantic-sensitive zones without storing raw spans. The counts intentionally over-approximate. They are policy signals for later transform gates, not a parser. Metadata must never include the matched path, identifier, hash, or string contents because receipts are safe to share. """ lines = text.splitlines() fence_markers = len(CODE_FENCE_RE.findall(text)) diff_lines = sum( 1 for line in lines if DIFF_FILE_HEADER_RE.match(line) or DIFF_HUNK_RE.match(line) or (line[:1] in "+-" and not line.startswith(("+++", "---"))) ) counts = { "code_fence": (fence_markers + 1) // 2, "diff": diff_lines, "identifier": len(IDENTIFIER_RE.findall(text)), "numeric_constant": len(NUMERIC_CONSTANT_RE.findall(text)), "hash": len(HASH_RE.findall(text)), "path": len(PATH_RE.findall(text)), "stack_frame": len(STACK_FRAME_RE.findall(text)), "quoted_string": len(QUOTED_STRING_RE.findall(text)), "json_key": len(JSON_KEY_RE.findall(text)), } return {key: counts[key] for key in PROTECTED_ZONE_KEYS if counts.get(key, 0) > 0} def build_protected_policy( *, text: str, content_type: str, strategy_detail: dict[str, object], lossy: bool, ) -> dict[str, object]: """Build an opt-in transform policy for protected zones. Protection governs transform eligibility and exact-retrieval expectations. It does not claim the section should be provider-cache-stable; cache ordering is handled by `context-guard-cost compile`. """ zone_counts = protected_zone_counts(text) detected = bool(zone_counts) strategy = str(strategy_detail.get("strategy") or "unknown") retrieval_required = bool(detected and lossy) return { "enabled": True, "detected": detected, "content_type": content_type, "zone_counts": zone_counts, "semantic_compress": False, "allowed_transforms": list(PROTECTED_ALLOWED_TRANSFORMS), "denied_transforms": list(PROTECTED_DENIED_TRANSFORMS), "retrieval_required": retrieval_required, "retrieval_scope": "sanitized_full_input" if retrieval_required else "compressed_output", "raw_spans_stored": False, "policy_note": "Protected zones permit structural transforms only; no semantic/paraphrase rewrites.", "strategy": { "name": strategy, "structural_only": True, }, } def build_transform_policy(protected_policy: dict[str, object]) -> dict[str, object]: """Summarize transform eligibility without embedding raw protected content.""" return { "mode": "protected" if protected_policy.get("detected") else "structural_default", "semantic_transforms_allowed": False, "semantic_compress": False, "allowed": list(PROTECTED_ALLOWED_TRANSFORMS), "denied": list(PROTECTED_DENIED_TRANSFORMS), "exact_retrieval_required": bool(protected_policy.get("retrieval_required")), "raw_spans_stored": False, } def build_readable_compression_metadata( *, content_type: str, strategy_detail: dict[str, object], lossy: bool, ) -> dict[str, object]: blocking = strategy_detail.get("readable_blocking_signals", {}) if not isinstance(blocking, dict): blocking = {} applied = bool(strategy_detail.get("readable_applied")) exact_fallback_required = bool(lossy or applied) return { "schema_version": READABLE_COMPRESSION_SCHEMA_VERSION, "mode": "readable", "preview_only": True, "applied": applied, "content_type": content_type, "strategy": strategy_detail.get("strategy"), "readable_strategy": strategy_detail.get("readable_strategy", "structural-preview"), "omitted_reason": strategy_detail.get("readable_omitted_reason"), "blocking_signal_counts": blocking, "protected_spans_stored": False, "source_verification": { "exact_fallback_required": exact_fallback_required, "recommended_command": "context-guard-artifact store --command 'readable-mode exact fallback' --json < sanitized-prose.txt", "verify_before_edit_or_claim": True, }, "claim_boundary": { "deterministic_local_only": True, "no_network_model_embedding_or_reranker": True, "no_generated_semantic_rewrite": True, "byte_and_token_counts_are_local_proxies": True, "hosted_api_token_or_cost_savings_claim_allowed": False, }, } def parse_json_candidate(stripped: str) -> object: if not stripped or stripped[0] not in "{[": return JSON_PARSE_FAILED try: return json.loads(stripped) except (ValueError, RecursionError): return JSON_PARSE_FAILED def _looks_like_json(stripped: str) -> bool: return parse_json_candidate(stripped) is not JSON_PARSE_FAILED def _ratio(matches: int, total: int, threshold: float) -> bool: return bool(total) and (matches / total) >= threshold def _looks_like_diff(sample: list[str]) -> bool: headers = sum(1 for line in sample if DIFF_FILE_HEADER_RE.match(line) or DIFF_HUNK_RE.match(line)) changes = sum(1 for line in sample if line[:1] in "+-" and not line.startswith(("+++", "---"))) return headers >= 1 and (changes >= 1 or headers >= 2) def _looks_like_search(sample: list[str]) -> bool: matches = sum(1 for line in sample if SEARCH_LINE_RE.match(line)) return _ratio(matches, len(sample), 0.6) and len(sample) >= 2 def _looks_like_log(sample: list[str]) -> bool: matches = sum(1 for line in sample if LOG_TIMESTAMP_RE.match(line) or LOG_LEVEL_RE.search(line)) return _ratio(matches, len(sample), 0.4) def _looks_like_code(sample: list[str]) -> bool: matches = sum(1 for line in sample if CODE_SIGNAL_RE.search(line)) return _ratio(matches, len(sample), 0.25) def compress_parsed_json(text: str, parsed: object) -> tuple[str, dict[str, object]]: compact = json.dumps(parsed, ensure_ascii=False, separators=(",", ":")) if not text.endswith("\n"): trailing = "" else: trailing = "\n" return compact + trailing, {"strategy": "json-compact", "lossy": False, "json_parse_ok": True} def compress_json(text: str) -> tuple[str, dict[str, object]]: """Re-serialize JSON without insignificant whitespace (data-preserving).""" parsed = parse_json_candidate(text.strip()) if parsed is JSON_PARSE_FAILED: # 파싱 불가 시 무손실을 깨지 않도록 prose 전략으로 안전하게 폴백한다. compressed, detail = compress_prose(text) detail["fallback_from"] = "json" return compressed, detail return compress_parsed_json(text, parsed) def compress_diff(text: str) -> tuple[str, dict[str, object]]: """Keep file headers, hunk headers, and +/- changes; collapse context runs.""" out: list[str] = [] context_run = 0 collapsed = 0 def flush() -> None: nonlocal context_run, collapsed if context_run: out.append(f"[context-guard-kit] {context_run} unchanged context line(s) omitted") collapsed += context_run context_run = 0 for line in text.splitlines(): is_structural = bool(DIFF_FILE_HEADER_RE.match(line) or DIFF_HUNK_RE.match(line)) is_change = line[:1] in "+-" and not line.startswith(("+++", "---")) if is_structural or is_change: flush() out.append(line) elif line.startswith(" ") or line == "": context_run += 1 else: flush() out.append(line) flush() return _join_lines(out, text), {"strategy": "diff-keep-changes", "lossy": True, "context_lines_omitted": collapsed} def compress_log(text: str) -> tuple[str, dict[str, object]]: """Collapse consecutive identical lines into a single `line (xN)` marker.""" out: list[str] = [] collapsed = 0 previous: str | None = None run = 0 def flush() -> None: nonlocal previous, run, collapsed if previous is None: return if run > 1: out.append(f"{previous} (x{run})") collapsed += run - 1 else: out.append(previous) previous, run = None, 0 for line in text.splitlines(): if line == previous: run += 1 continue flush() previous, run = line, 1 flush() return _join_lines(out, text), {"strategy": "log-collapse-repeats", "lossy": True, "lines_collapsed": collapsed} def compress_search(text: str) -> tuple[str, dict[str, object]]: """Drop exact-duplicate match lines while preserving first-seen order with bounded keys.""" out: list[str] = [] seen: set[str] = set() dropped = 0 dedupe_limit_reached = False for line in iter_text_lines(text): key = line.rstrip() if key in seen: dropped += 1 continue if len(seen) < MAX_SEARCH_DEDUPE_KEYS: seen.add(key) else: dedupe_limit_reached = True out.append(line) return _join_lines(out, text), { "strategy": "search-dedupe", "lossy": dropped > 0, "duplicate_lines_dropped": dropped, "dedupe_key_limit": MAX_SEARCH_DEDUPE_KEYS, "dedupe_key_limit_reached": dedupe_limit_reached, } def compress_code(text: str) -> tuple[str, dict[str, object]]: """Trim trailing whitespace and collapse 3+ blank lines to a single blank.""" return _whitespace_normalize(text, strategy="code-whitespace", max_consecutive_blank=1) def compress_prose(text: str) -> tuple[str, dict[str, object]]: """Trim trailing whitespace and collapse 2+ blank lines to a single blank.""" return _whitespace_normalize(text, strategy="prose-whitespace", max_consecutive_blank=1) def readable_blocking_signal_counts(text: str, content_type: str) -> dict[str, int]: counts = protected_zone_counts(text) blocking = { key: int(counts.get(key, 0) or 0) for key in READABLE_BLOCKING_PROTECTED_KEYS if int(counts.get(key, 0) or 0) > 0 } prompt_like = len(PROMPT_LIKE_INSTRUCTION_RE.findall(text)) if prompt_like: blocking["prompt_like_instruction"] = prompt_like if content_type != "prose": blocking["non_prose_content"] = 1 return blocking def split_prose_sentences(text: str) -> list[str]: compact = " ".join(text.split()) if not compact: return [] sentences = re.split(r"(?<=[.!?])\s+", compact) return [sentence.strip() for sentence in sentences if sentence.strip()] def compress_prose_readable(text: str) -> tuple[str, dict[str, object]]: """Readable opt-in sentence window for sanitized unprotected prose only.""" normalized, base_detail = compress_prose(text) blocking = readable_blocking_signal_counts(normalized, "prose") detail = dict(base_detail) detail.update({ "readable_mode": True, "readable_strategy": "sentence-window-preview", "readable_blocking_signals": blocking, }) if blocking: detail["readable_applied"] = False detail["readable_omitted_reason"] = "protected_or_prompt_like_signal" return normalized, detail sentences = split_prose_sentences(normalized) if len(sentences) <= READABLE_SENTENCE_LIMIT: detail["readable_applied"] = False detail["readable_omitted_reason"] = "short_prose" return normalized, detail included_sentences = sentences[:3] + sentences[-1:] kept = sentences[:3] + [f"[context-guard-readable] {len(sentences) - len(included_sentences)} sentence(s) omitted; retrieve exact source before relying on omitted detail."] + sentences[-1:] preview = " ".join(kept) if text.endswith("\n"): preview += "\n" detail.update({ "strategy": "prose-readable-window", "lossy": True, "readable_applied": True, "sentences_original": len(sentences), "sentences_included": len(included_sentences), "sentences_omitted": len(sentences) - len(included_sentences), }) return preview, detail def _whitespace_normalize(text: str, *, strategy: str, max_consecutive_blank: int) -> tuple[str, dict[str, object]]: out: list[str] = [] blank_run = 0 collapsed = 0 for line in text.splitlines(): trimmed = line.rstrip() if trimmed == "": blank_run += 1 if blank_run > max_consecutive_blank: collapsed += 1 continue else: blank_run = 0 out.append(trimmed) lossy = collapsed > 0 or any(line != line.rstrip() for line in text.splitlines()) return _join_lines(out, text), {"strategy": strategy, "lossy": lossy, "blank_lines_collapsed": collapsed} def _join_lines(lines: list[str], original: str) -> str: """Join compressed lines, restoring a trailing newline only if the input had one.""" body = "\n".join(lines) if original.endswith("\n") and body and not body.endswith("\n"): body += "\n" return body STRATEGIES: dict[str, Callable[[str], tuple[str, dict[str, object]]]] = { "json": compress_json, "diff": compress_diff, "log": compress_log, "search": compress_search, "code": compress_code, "prose": compress_prose, } def build_metadata( *, content_type: str, type_source: str, strategy_detail: dict[str, object], original_text: str, compressed_text: str, redacted_lines: int, input_truncated: bool, input_bytes: int, max_bytes: int, protected_policy_enabled: bool = False, compression_mode: str = "conservative", ) -> dict[str, object]: """Assemble the compress receipt: observed byte/line counts plus an estimated token proxy. `redacted_lines` is computed before this point (redaction-before-receipt), so the metadata can be safely emitted. A deterministic retrieval hint points at escrow for exact-byte recovery because every strategy except json-compact is lossy. """ original_bytes = byte_length(original_text) compressed_bytes = byte_length(compressed_text) ratio = round(compressed_bytes / original_bytes, 4) if original_bytes else 1.0 lossy = bool(strategy_detail.get("lossy", True)) retrieval_hint = ( "Lossy: store the full sanitized text for exact recovery via " "`context-guard-artifact store` and query slices later." if lossy else "Data-preserving: compact form is semantically equivalent to the sanitized input." ) metadata: dict[str, object] = { "tool": "context-guard-kit.context_compress", "metadata_version": 1, "content_type": content_type, "type_source": type_source, "strategy": strategy_detail.get("strategy"), "strategy_detail": strategy_detail, "lossy": lossy, "input": { "bytes_read": input_bytes, "truncated": input_truncated, "max_bytes": max_bytes, }, "redaction": { "redacted_lines": redacted_lines, "redacted_before_receipt": True, }, "bytes": { "measurement": "observed", "original": original_bytes, "compressed": compressed_bytes, "saved": original_bytes - compressed_bytes, "compression_ratio": ratio, }, "lines": { "measurement": "observed", "original": line_count(original_text), "compressed": line_count(compressed_text), }, "token_proxy": { "measurement": "estimated", "method": f"chars_div_{TOKEN_PROXY_CHARS_PER_TOKEN}", "original": token_proxy(original_text), "compressed": token_proxy(compressed_text), }, "retrieval_hint": retrieval_hint, } if protected_policy_enabled: protected_policy = build_protected_policy( text=original_text, content_type=content_type, strategy_detail=strategy_detail, lossy=lossy, ) metadata["protected_zone_policy"] = protected_policy metadata["transform_policy"] = build_transform_policy(protected_policy) if protected_policy.get("retrieval_required"): metadata["retrieval_hint"] = ( "Protected lossy structural transform: store the full sanitized text with " "`context-guard-artifact store` and retrieve exact slices before relying on omitted content." ) if compression_mode == "readable": metadata["readable_compression"] = build_readable_compression_metadata( content_type=content_type, strategy_detail=strategy_detail, lossy=lossy, ) return metadata def compress_text( text: str, *, forced_type: str | None, show_paths: bool, input_truncated: bool, input_bytes: int, max_bytes: int, protected_policy_enabled: bool = False, compression_mode: str = "conservative", sanitization_context: str = "unknown_text", private_roots: tuple[str, ...] = (), ) -> tuple[str, dict[str, object]]: """Sanitize first, then classify and compress, then build the receipt. Redaction runs on the raw input so no secret can leak into the classifier, the compressed body, or the metadata that follows. """ sanitized, redacted_lines = sanitize_text( text, show_paths=show_paths, context=sanitization_context, private_roots=private_roots, ) parsed_json: object = JSON_PARSE_FAILED if forced_type is not None: content_type, type_source = forced_type, "override" else: stripped = sanitized.strip() parsed_json = parse_json_candidate(stripped) content_type = "json" if parsed_json is not JSON_PARSE_FAILED else classify_non_json_content(stripped) type_source = "detected" if compression_mode == "readable" and content_type == "prose": compressed, strategy_detail = compress_prose_readable(sanitized) else: if content_type == "json" and parsed_json is not JSON_PARSE_FAILED: compressed, strategy_detail = compress_parsed_json(sanitized, parsed_json) else: compressed, strategy_detail = STRATEGIES[content_type](sanitized) if compression_mode == "readable": strategy_detail["readable_mode"] = True strategy_detail["readable_strategy"] = "sentence-window-preview" strategy_detail["readable_applied"] = False strategy_detail["readable_omitted_reason"] = "non_prose_content" strategy_detail["readable_blocking_signals"] = {"non_prose_content": 1} # 보수성 보장: 어떤 전략도 입력보다 큰 결과를 내보내지 않는다. 작은 입력에서 # 접기 마커가 원본보다 길어지는 경우 살균된 원본을 그대로 유지한다. if byte_length(compressed) >= byte_length(sanitized): compressed = sanitized if compression_mode == "readable" and strategy_detail.get("readable_applied"): strategy_detail["lossy"] = False strategy_detail["readable_applied"] = False strategy_detail["readable_omitted_reason"] = "not_smaller_than_input" strategy_detail["reduced"] = False else: strategy_detail["reduced"] = True metadata = build_metadata( content_type=content_type, type_source=type_source, strategy_detail=strategy_detail, original_text=sanitized, compressed_text=compressed, redacted_lines=redacted_lines, input_truncated=input_truncated, input_bytes=input_bytes, max_bytes=max_bytes, protected_policy_enabled=protected_policy_enabled, compression_mode=compression_mode, ) redaction_metadata = metadata.get("redaction") if isinstance(redaction_metadata, dict): redaction_metadata["context"] = sanitization_context return compressed, metadata def render_text_receipt(metadata: dict[str, object]) -> str: """One-block human summary written to stderr in text mode.""" byte_stats = metadata.get("bytes", {}) token_stats = metadata.get("token_proxy", {}) redaction = metadata.get("redaction", {}) lines = [ "[context-guard-kit] compress", f"- content_type: {metadata.get('content_type')} ({metadata.get('type_source')})", f"- strategy: {metadata.get('strategy')} (lossy={str(metadata.get('lossy')).lower()})", ] if isinstance(byte_stats, dict): lines.append( f"- bytes: {byte_stats.get('original')} -> {byte_stats.get('compressed')} " f"(ratio={byte_stats.get('compression_ratio')})" ) if isinstance(token_stats, dict): lines.append( f"- token_proxy(estimated): {token_stats.get('original')} -> {token_stats.get('compressed')}" ) if isinstance(redaction, dict) and redaction.get("redacted_lines"): lines.append(f"- redacted_lines: {redaction.get('redacted_lines')}") return "\n".join(lines) + "\n" def run_compress(args: argparse.Namespace) -> int: """Read stdin, compress, then emit JSON or (compressed text + stderr receipt).""" max_bytes = bounded_int(args.max_bytes, DEFAULT_MAX_BYTES, 1, MAX_MAX_BYTES) compression_mode = args.mode if compression_mode not in COMPRESSION_MODES: print(f"context-guard-compress: unknown --mode: {compression_mode}", file=sys.stderr) return 2 raw_text, input_truncated, input_bytes = read_bounded_stdin(max_bytes) forced_type = args.type if forced_type is not None and forced_type not in STRATEGIES: print(f"context-guard-compress: unknown --type: {forced_type}", file=sys.stderr) return 2 compressed, metadata = compress_text( raw_text, forced_type=forced_type, show_paths=args.show_paths, input_truncated=input_truncated, input_bytes=input_bytes, max_bytes=max_bytes, protected_policy_enabled=bool(args.protected_policy), compression_mode=compression_mode, sanitization_context=( "source_code" if forced_type == "code" else args.sanitize_context ), private_roots=tuple(args.private_root), ) if args.json: payload = {"metadata": metadata, "content": compressed} print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) elif args.metadata_only: print(json.dumps(metadata, ensure_ascii=False, indent=2, sort_keys=True)) else: sys.stdout.write(compressed) if not args.quiet: sys.stderr.write(render_text_receipt(metadata)) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="[deprecated] Classify and conservatively compress stdin (sanitized) for token-budget reuse. Built-in compaction wins in practice; use context-guard-artifact for lossless retrieval instead.", ) parser.add_argument( "--type", choices=CONTENT_TYPES, default=None, help="force a content type instead of auto-detecting (json/diff/log/search/code/prose)", ) parser.add_argument( "--mode", choices=COMPRESSION_MODES, default="conservative", help="compression policy: conservative keeps existing deterministic strategies; readable adds opt-in readable preview/source-verification metadata", ) parser.add_argument("--json", action="store_true", help="emit JSON with metadata and compressed content") parser.add_argument( "--protected-policy", action="store_true", help="add opt-in protected-zone transform policy metadata to --json/--metadata-only receipts; default content is unchanged", ) parser.add_argument( "--metadata-only", action="store_true", help="emit only the JSON metadata receipt (no compressed body)", ) parser.add_argument("--quiet", action="store_true", help="suppress the text receipt on stderr in text mode") parser.add_argument( "--show-paths", action="store_true", help="show raw absolute paths instead of path hashes; local debugging only because private paths may be exposed", ) parser.add_argument( "--sanitize-context", choices=( "unknown_text", "command_search_diff", "filesystem_listing", "source_code", ), default="unknown_text", help="declare the input origin for conservative secret/path sanitization", ) parser.add_argument( "--private-root", action="append", default=[], help="private root for filesystem_listing sanitization; may be repeated", ) parser.add_argument("--max-bytes", type=int, default=DEFAULT_MAX_BYTES, help="maximum stdin bytes to read before truncating") parser.set_defaults(func=run_compress) return parser def main() -> int: parser = build_parser() args = parser.parse_args() try: return int(args.func(args)) except RuntimeError as exc: print(f"context-guard-compress: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())