QA and context compress

docs/plans/CASAN_BACKLOG_STATUS.md

@@ -57,7 +57,7 @@
 | **04 Self-improve** | � core done+test | `packages/casan-harness/scripts/bash/self-improve.py`: `propose` đọc metrics/drift → proposal dry-run (không ghi); `apply` bắt buộc approval, áp qua governed store (audit); sensitive/loosen luôn cần duyệt. `phase-selfimprove-tests.sh` 7/0 (WSL), nối CI. Còn: luật đề xuất phong phú hơn (corpus/model escalation), chạy định kỳ CI (05). |
 | **05 CI/CD** | 🟡 CI gate MVP done | `packages/casan-harness/scripts/bash/ci-harness-gate.sh` chạy các suite harness/hardening/sourcegen/traceability/frontend theo thứ tự an toàn, có timeout/filter; `.gitea/workflows/harness-ci.yml` gọi gate trên push/PR. Filtered local verify PASS=2/0. Còn: full gate xanh trên runner thật, xử lý A6 nếu còn chậm/treo, bật Docker infra lab nếu runner hỗ trợ, package/release artifact `fpt-casan-sdd-harness`. |
 | **06 Onboard dự án 2** | 📋 chưa bắt đầu (🔓 **đã mở khoá** — 01 done) | Chứng minh reuse: cắm 1 repo khác + golden/corpus/input, đăng ký qua `verify-harness-reuse.sh`, không sửa gate. Phụ thuộc 01 ✅. App mới chỉ cần `apps/<project>/domain/` + set `CASAN_DOMAIN_ROOT` (đã có `domain_root` per-project trong `project-registry.json`). |
-| **08 Context compression** | � MVP done+test | **CASAN-native token-killer** (Track 3) đã có: `context-compress.py` (dedup/extractive/structural, must-keep, tee, gate fail-able), `phase08-compression-tests.sh` 7/0 (WSL), nối CI. Còn: Track 1 nén INPUT + Track 2 nén VIEW liên-bước + Track 4 abstractive (gated) + nối H4/H5 trong pipeline thật. |
+| **08 Context compression** | 🟡 Track 3 hardened+test | **CASAN-native token-killer** đã có: severity `[ERR]/[WRN]`, multiplicity-preserving must-keep gate, raw fallback/halt fail-safe, hash-bound JSON evidence, estimate/provider labeling; `phase08-compression-tests.sh` 19/0, SEC-09 7/0, loop consumer 16/0. Log thật NEHOPS giữ 4/4 warning; 1.620→76 chỉ là whitespace estimate, chưa phải ROI claim. Còn: Track 1 INPUT + Track 2 VIEW + Track 4 abstractive + H4/H5 pipeline binding + A/B provider telemetry. |
 | **12 Domain Pack SDK** | 📋 chưa bắt đầu (🔓 01 done, còn chờ 06) | Onboard bằng khai báo (golden/corpus/policy theo domain). Phụ thuộc 01 ✅, 06. |
 | **01 Restructure** | ✅ **done+test (2026-07-08, merged main)** | Đã tách: harness code → `packages/casan-harness/`, domain OKR → `apps/okr/domain/`, runtime state ở `.specify/`; **app promote lên git root** (hết wrapper `AINative_OKR_CASAN5/`); facade `.specify` symlink **gỡ sạch** (hard cutoff); path resolve qua `casan-paths.sh` (marker walk-up). Full gate **PASS=64 FAIL=0 SKIP=3** từ cấu trúc mới. CI (`.gitea/workflows/ci.yml`) + docs đã đồng bộ. Mở khoá 06/12. Chi tiết: `CASAN_PLAN_01_RESTRUCTURE.md`. |
 | **13 Control Plane** | 🟡 core + **Track 1/2/3 + FinOps/SLO + Command Center + local-prod TLS/OIDC smoke** done+test | **Sửa kiến trúc: là tài sản harness, KHÔNG nằm trong OKR.** Governance core đã dời vào harness: `packages/casan-harness/scripts/bash/control-plane-settings.py` (settings versioned + audit hash-chain + deny-by-default + approval + rollback), `phase-control-plane-tests.sh` 9/0 + `phase-control-plane-hitl-tests.sh` 9/0 (nối CI). Đã gỡ khỏi `apps/okr` (OKR sạch: 46/0/3skip + 16/16). **Track 1 DONE**: Ops Console read-only telemetry. **Track 2 DONE**: settings API/UI tại `packages/casan-control-panel/` bọc `control-plane-settings.py` + `rbac-check.py`. **Track 3 DONE**: kill-switch API/UI; approval inbox/delegation/oversight API/UI bọc harness `approval-inbox.py` + `delegation-policy.yaml`, SoD, apply setting proposal qua governed store; IdP group claim→RBAC role mapping tested. **Track 4 local-prod DONE**: FinOps/SLO page + Docker/nginx/oauth2-proxy deploy scaffold; `local-prod-smoke.sh` passes `CP_LOCAL_SMOKE_PASS https_oidc=true actor=oidc-ops role=org-admin` and invokes authenticated `managed-prod-smoke.sh` on the local-prod endpoint (`CP_MANAGED_SMOKE_PASS actor=oidc-ops role=org-admin widgets=9`); `prod-readiness-check.sh` validates managed-prod TLS/OIDC/nginx prerequisites. **Command Center §8.6 DONE**: `GET /api/v1/command` + `/command` UI, 9 evidence-backed widgets (gồm `chat_loop`), provenance envelope, executive briefing VI/EN, live ticker, evidence drawer. `npm run console:test` **26/0**, `npm run console:build` xanh. Targeted related suites: RBAC 12/0, C7 15/0. Còn: live managed prod deploy với cert/enterprise IdP/host thật (07 T2); RAI view thuộc Plan-15 follow-up. |

docs/plans/CASAN_PLAN_08_CONTEXT_COMPRESSION.md

@@ -2,13 +2,19 @@
 
 > Năng lực **nén prompt/context để giảm token · latency · cost** — nhưng **không phá governance**.
 >
-> Status 2026-07-06: **🟡 Track 3 MVP implemented + tested (CASAN-native token-killer).**
+> Status 2026-07-29: **🟡 Track 3 hardened + regression-tested (CASAN-native token-killer).**
 > `packages/casan-harness/scripts/bash/context-compress.py` — compressor deterministic của CASAN
-> (KHÔNG dùng lại RTK): modes `dedup`/`extractive`/`structural`, must-keep luôn giữ,
+> (KHÔNG dùng lại RTK): modes `dedup`/`extractive`/`structural`; nhận diện severity
-> tee raw-passthrough khi lệnh fail, báo `token_saved`/`ratio`, và gate must-keep
+> `[ERR]/[WRN]`; giữ multiplicity của warning/must-keep; invalid regex/input rỗng sau nén
-> fail-able. Test `packages/casan-harness/tests/phase08-compression-tests.sh` **7/0 (WSL)**; đã nối
+> đều fail-closed; preservation fail mặc định trả RAW + exit 1; có lựa chọn `halt`; JSON
-> vào `ci-harness-gate.sh` (`phase08-compression`). Còn: Track 1 nén INPUT + Track 2
+> evidence bind SHA-256 raw/candidate/output và ghi rõ `whitespace_estimate` khác provider
-> nén VIEW liên-bước + Track 4 abstractive (gated) + nối H4/H5 scan/hash trong pipeline thật.
+> telemetry. Test `packages/casan-harness/tests/phase08-compression-tests.sh` **19/0 (macOS)**;
+> SEC-09 **7/0**; loop consumer **16/0**; đã nối `ci-harness-gate.sh`
+> (`phase08-compression`). Log thật NEHOPS `analyzer-20260507.log` đo **1.620 → 76
+> whitespace-estimate units**, giữ đủ **4/4 `[WRN]`**, preservation pass; con số này vẫn
+> **không phải provider token saving claim** cho đến khi qua H3 end-to-end/A-B. Còn:
+> Track 1 nén INPUT + Track 2 nén VIEW liên-bước + Track 4 abstractive (gated) + nối
+> H4 scan raw/compressed và H5 audit binding trong pipeline thật.
 >
 > Nhãn: [có] tồn tại thật · [đo] đã kiểm chứng · [mới] cần làm · [chưa tự động] có đo/người quyết.
 > Phụ thuộc: **01** (đường dẫn sau restructure) · **07** (thứ tự scan/audit an toàn, V19/V5) · **03** (chế độ abstractive dùng model).

packages/casan-harness/scripts/bash/context-compress.py

@@ -1,45 +1,62 @@
 #!/usr/bin/env python3
-"""CASAN-native token-killer (Plan-08 Track 3).
+"""CASAN-native deterministic context compressor (Plan-08 Track 3).
 
-A deterministic tool-output compressor written for CASAN — NOT a wrapper around
+The compressor reduces long tool output before it enters model context while
-RTK. It reduces token count of long command/tool output before it enters model
+preserving operationally significant lines. It is deliberately not a tokenizer
-context, while (a) always preserving must-keep lines, (b) never compressing on
+or a billing source: token figures are whitespace-based estimates and are
-failure (raw passthrough for debugging, RTK-style tee), and (c) reporting the
+labelled as such.
-token savings for H6 telemetry.
 
-Modes:
+Safety contract:
-  dedup       collapse consecutive duplicate lines with an (xN) counter
+  * preserve severity-tagged errors/warnings, failure/skip signals, summaries,
-  extractive  keep only important lines (errors/failures/warnings) + must-keep
+    operational result lines, and project-supplied must-keep patterns;
-  structural  dedup + keep summary/important/must-keep lines (for test/log output)
+  * 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 note: this runs AFTER `H4 scan raw` + `H5 hash raw` and BEFORE
+Governance ordering remains the caller's responsibility:
-`H4 scan compressed` in the Plan-08 pipeline; it is deterministic and needs no
+  H4 scan raw -> H5 hash raw -> compress -> H3 preservation/faithfulness
-model, so it cannot be used as a path to evade H4.
+  -> H4 scan compressed -> H5 bind raw/compressed hashes -> model.
 """
+
 import argparse
+from collections import Counter
+import hashlib
 import json
 import os
-
-
-def _casan_app_root():
-    # Plan-01: walk UP for the `.specify` state marker (harness code lives in
-    # packages/casan-harness/, so a fixed __file__ depth would mis-root).
-    _d = os.path.abspath(os.path.dirname(__file__))
-    _p = _d
-    while _p != os.path.dirname(_p):
-        if os.path.isdir(os.path.join(_p, ".specify")) or os.path.isdir(os.path.join(_p, "packages/casan-harness")):
-            return _p
-        _p = os.path.dirname(_p)
-    return os.path.abspath(os.path.join(_d, "..", "..", ".."))
-
 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:
-    """Read the effective `compression.enabled` from the control-plane settings
+    """Return the governed ``compression.enabled`` setting.
-    store (the harness-owned governed settings). Absent/invalid ⇒ enabled (default).
+
-    This is how a Control Plane setting change actually governs the harness."""
+    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(
@@ -50,145 +67,544 @@ def compression_enabled() -> bool:
     if not os.path.isfile(store_file):
         return True
     try:
-        data = json.load(open(store_file, encoding="utf-8"))
+        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
 
 
-IMPORTANT_RE = re.compile(
+# Serilog/log4net-style severity tags need explicit recognition. Word-boundary
-    r"\b(error|errors|fail|failed|failure|failing|exception|panic|denied|blocked|warn|warning)\b",
+# 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,
 )
-SUMMARY_RE = re.compile(r"\b(\d+)\s+(pass|passed|fail|failed|tests?|errors?|warnings?)\b", 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 estimate_tokens(text: str) -> int:
+def configured_max_bytes() -> int:
-    return len(text.split())
+    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
 
 
-# SEC-09 (M-10): bound input size (DoS) and read fail-closed. Non-UTF8 degrades via
+def read_capped(src: str, max_bytes: int) -> Tuple[str, int]:
-# errors="replace" instead of crashing; oversize/unreadable input exits non-zero and
+    """Read at most ``max_bytes`` and return text plus replacement count."""
-# emits nothing (never a crash traceback, never silent truncation).
-MAX_BYTES = int(os.environ.get("CASAN_MAX_INPUT_BYTES", str(2 * 1024 * 1024)))
-
-
-def read_capped(src: str) -> str:
     try:
         if src == "-":
-            data = sys.stdin.buffer.read(MAX_BYTES + 1)
+            data = sys.stdin.buffer.read(max_bytes + 1)
         else:
-            with open(src, "rb") as fh:
+            with open(src, "rb") as handle:
-                data = fh.read(MAX_BYTES + 1)
+                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:
+    if len(data) > max_bytes:
-        print(f"COMPRESS_FAIL input_exceeds_cap({MAX_BYTES}B) fail-closed", file=sys.stderr)
+        print(
+            f"COMPRESS_FAIL input_exceeds_cap({max_bytes}B) fail-closed",
+            file=sys.stderr,
+        )
         raise SystemExit(1)
-    return data.decode("utf-8", errors="replace")
+    text = data.decode("utf-8", errors="replace")
+    return text, text.count("\ufffd")
 
 
-def load_patterns(path: str):
+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 fh:
+        with open(path, encoding="utf-8", errors="replace") as handle:
-            return [line.strip() for line in fh if line.strip()]
+            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] = []
-def is_must_keep(line: str, patterns) -> bool:
+    for line_number, expression in raw_rules:
-    return any(re.search(p, line) for p in patterns)
+        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 dedup(lines):
+def is_must_keep(line: str, patterns: Sequence[PatternRule]) -> bool:
-    out = []
+    return any(pattern.search(line) for _, pattern in patterns)
-    i = 0
-    n = len(lines)
-    while i < n:
-        j = i
-        while j + 1 < n and lines[j + 1] == lines[i]:
-            j += 1
-        count = j - i + 1
-        out.append(lines[i] if count == 1 else f"{lines[i]}  (x{count})")
-        i = j + 1
-    return out
 
 
-def compress(text: str, mode: str, must):
+def is_protected(line: str, patterns: Sequence[PatternRule]) -> bool:
-    lines = text.split("\n")
+    """Return whether a line is forbidden from disappearing."""
-    if mode == "dedup":
+    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 [ln for ln in lines if IMPORTANT_RE.search(ln) or is_must_keep(ln, must)]
+        return lines
-    if mode == "structural":
-        kept = [
-            ln
-            for ln in lines
-            if IMPORTANT_RE.search(ln) or SUMMARY_RE.search(ln) or is_must_keep(ln, must)
-        ]
-        return dedup(kept)
     raise ValueError(f"unknown mode: {mode}")
 
 
-def main() -> int:
+def compression_source_lines(
-    ap = argparse.ArgumentParser()
+    raw_lines: Sequence[str],
-    ap.add_argument("--mode", choices=["dedup", "extractive", "structural"], default="structural")
+    mode: str,
-    ap.add_argument("--input", default="-", help="input file or - for stdin")
+    must_keep: Sequence[PatternRule],
-    ap.add_argument("--must-keep-file", default="", help="file with one must-keep regex per line")
+) -> List[str]:
-    ap.add_argument("--failed", action="store_true", help="raw passthrough (tee) when the command failed")
+    """Return the exact raw lines from which a candidate may be built."""
-    ap.add_argument(
+    if mode == "dedup":
-        "--respect-policy",
+        return list(raw_lines)
-        action="store_true",
+    return [line for line in raw_lines if is_protected(line, must_keep)]
-        help="honor control-plane `compression.enabled`; if disabled, pass raw through",
-    )
-    ap.add_argument(
-        "--require-must-keep-file",
-        default="",
-        help="verify every pattern in this file still appears; exit 1 (gate) if any is missing",
-    )
-    args = ap.parse_args()
 
-    raw = read_capped(args.input)
-    must = load_patterns(args.must_keep_file)
 
-    if args.failed:
+def expanded_line_counts(
-        # RTK-style tee: never compress failing output; keep raw for debugging.
+    lines: Iterable[str],
-        out_text = raw
+    dedup_encoded: bool,
-        mode_used = "passthrough"
+    source_lines: Sequence[str],
-    elif args.respect_policy and not compression_enabled():
+) -> Counter:
-        # Control-plane setting governs the harness: compression disabled ⇒ raw.
+    """Decode ``(xN)`` markers produced by ``dedup`` into weighted counts."""
-        out_text = raw
+    candidate_lines = list(lines)
-        mode_used = "policy-disabled"
+    # 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:
-        out_lines = compress(raw, args.mode, must)
+            counts[line] += 1
-        out_text = "\n".join(out_lines)
+    return counts
-        mode_used = args.mode
 
-    in_tokens = estimate_tokens(raw)
-    out_tokens = estimate_tokens(out_text)
-    saved = in_tokens - out_tokens
-    ratio = round(out_tokens / in_tokens, 4) if in_tokens else 1.0
 
-    verify_patterns = load_patterns(args.require_must_keep_file)
+def protected_line_deficits(
-    missing = [p for p in verify_patterns if not re.search(p, out_text)]
+    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
+    }
 
-    sys.stdout.write(out_text)
+
-    if not out_text.endswith("\n"):
+def required_pattern_deficits(
-        sys.stdout.write("\n")
+    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(
-        f"COMPRESS mode={mode_used} in_tokens={in_tokens} out_tokens={out_tokens} "
+            "COMPRESS_FAIL report_json_stdout_conflicts_with_compressed_output",
-        f"saved={saved} ratio={ratio} must_keep_missing={len(missing)}",
             file=sys.stderr,
         )
-    if missing:
+        raise SystemExit(1)
-        print(f"COMPRESS_MUST_KEEP_DROPPED {','.join(missing)}", file=sys.stderr)
+    absolute = os.path.abspath(path)
-        return 1
+    directory = os.path.dirname(absolute)
-    return 0
+    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__":

packages/casan-harness/tests/fixtures/context-compression/nehops/must-keep.patterns

@@ -0,0 +1,8 @@
+# NEHOPS analyzer invariants. These expressions are project/domain policy,
+# while generic severity-tag preservation belongs to the Core compressor.
+VBSource root not found
+common code tracing skipped
+Error Messages
+Output Directory
+Generated [0-9]+ files
+exit code

packages/casan-harness/tests/phase08-compression-tests.sh

@@ -10,6 +10,7 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 source "$SCRIPT_DIR/../scripts/bash/casan-paths.sh"
 PROJECT_ROOT="$CASAN_APP_ROOT"
 CC="$CASAN_HARNESS_ROOT/scripts/bash/context-compress.py"
+FIXTURES="$SCRIPT_DIR/fixtures/context-compression/nehops"
 WORK="$(mktemp -d)"
 trap 'rm -rf "$WORK"' EXIT
 
@@ -89,6 +90,178 @@ set -e 2>/dev/null || true
 [[ "$RC" -eq 0 ]] && pass "protecting the pattern keeps the line and passes the gate" \
   || fail "protected must-keep still failed the gate (rc=$RC)"
 
+# 8) Realistic NEHOPS regression: abbreviated [WRN] tags and domain invariants
+# survive structural compression, with hash-bound evidence that labels token
+# figures as whitespace estimates.
+set +e
+python3 "$CC" --mode structural \
+  --input "$FIXTURES/analyzer-sanitized.log" \
+  --must-keep-file "$FIXTURES/must-keep.patterns" \
+  --require-must-keep-file "$FIXTURES/must-keep.patterns" \
+  --report-json "$WORK/nehops-report.json" \
+  >"$WORK/nehops-compressed.log" 2>"$WORK/nehops.err"
+RC=$?
+set -e 2>/dev/null || true
+RAW_WARNINGS="$(grep -c '\[WRN\]' "$FIXTURES/analyzer-sanitized.log")"
+OUT_WARNINGS="$(grep -c '\[WRN\]' "$WORK/nehops-compressed.log" || true)"
+if [[ "$RC" -eq 0 && "$RAW_WARNINGS" -eq 2 && "$OUT_WARNINGS" -eq "$RAW_WARNINGS" ]] \
+  && grep -q "FRR04700_CL.*common code tracing skipped" "$WORK/nehops-compressed.log" \
+  && grep -q "FRR04700_SV.*common code tracing skipped" "$WORK/nehops-compressed.log" \
+  && grep -q "Error Messages: 0" "$WORK/nehops-compressed.log"; then
+  pass "NEHOPS [WRN] + CL/SV/domain invariants survive compression"
+else
+  fail "NEHOPS warning preservation failed (rc=$RC raw=$RAW_WARNINGS out=$OUT_WARNINGS)"
+fi
+
+set +e
+python3 - "$WORK/nehops-report.json" <<'PY'
+import json
+import sys
+
+report = json.load(open(sys.argv[1], encoding="utf-8"))
+assert report["decision"] == "compressed"
+assert report["preservation"]["status"] == "passed"
+assert report["preservation"]["protected_line_missing"] == 0
+assert report["preservation"]["required_pattern_deficits"] == []
+assert report["measurement_source"] == "whitespace_estimate"
+assert report["provider_telemetry"] is None
+assert report["saving_status"] == "estimate_only_quality_gate_required"
+assert report["severity"]["raw"]["warning"] == 2
+assert report["severity"]["candidate"]["warning"] == 2
+assert len(report["hashes"]["raw_sha256"]) == 64
+assert len(report["hashes"]["candidate_sha256"]) == 64
+PY
+RC=$?
+set -e 2>/dev/null || true
+[[ "$RC" -eq 0 ]] \
+  && pass "JSON evidence binds hashes and labels estimate/provider status" \
+  || fail "JSON evidence contract invalid"
+
+# 9) Required-pattern multiplicity: retaining one of two matching lines is not
+# sufficient. Rejection defaults to raw fallback and reports zero savings.
+printf 'TRACE-MUST plain line\nERROR TRACE-MUST protected line\n' > "$WORK/multi.txt"
+printf 'TRACE-MUST\n' > "$WORK/multi.patterns"
+set +e
+python3 "$CC" --mode extractive --input "$WORK/multi.txt" \
+  --require-must-keep-file "$WORK/multi.patterns" \
+  >"$WORK/multi.out" 2>"$WORK/multi.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 1 ]] \
+  && cmp -s "$WORK/multi.txt" "$WORK/multi.out" \
+  && grep -q "TRACE-MUST(1/2)" "$WORK/multi.err" \
+  && grep -q "saving_status=rejected" "$WORK/multi.err" \
+  && grep -q "saved=0" "$WORK/multi.err"; then
+  pass "partial must-keep loss is rejected with raw fallback and no saving claim"
+else
+  fail "multiplicity/fallback contract failed (rc=$RC)"
+fi
+
+# 10) A caller may choose halt instead of raw fallback. Halt returns non-zero
+# and emits no candidate bytes.
+set +e
+python3 "$CC" --mode extractive --input "$WORK/multi.txt" \
+  --require-must-keep-file "$WORK/multi.patterns" \
+  --on-preservation-failure halt \
+  >"$WORK/halt.out" 2>"$WORK/halt.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 1 && ! -s "$WORK/halt.out" ]] \
+  && grep -q "COMPRESS_REJECTED fallback=halt" "$WORK/halt.err" \
+  && grep -q "saving_status=rejected" "$WORK/halt.err" \
+  && grep -q "saved=0" "$WORK/halt.err"; then
+  pass "halt policy emits no lossy candidate on preservation failure"
+else
+  fail "halt policy leaked output or returned success (rc=$RC)"
+fi
+
+# 11) Required patterns absent from raw are not fabricated and do not create a
+# false failure; preservation checks only claims that raw evidence contained.
+printf 'ERROR real failure line\n' > "$WORK/absent.txt"
+printf 'NEVER_PRESENT_IN_RAW\n' > "$WORK/absent.patterns"
+set +e
+python3 "$CC" --mode extractive --input "$WORK/absent.txt" \
+  --require-must-keep-file "$WORK/absent.patterns" \
+  >"$WORK/absent.out" 2>"$WORK/absent.err"
+RC=$?
+set -e 2>/dev/null || true
+[[ "$RC" -eq 0 ]] \
+  && pass "pattern absent from raw does not create a false preservation failure" \
+  || fail "absent raw pattern incorrectly failed preservation (rc=$RC)"
+
+# 12) Invalid project regex is a configuration defect and fails closed without
+# emitting a candidate.
+printf '[unterminated\n' > "$WORK/invalid.patterns"
+set +e
+python3 "$CC" --mode structural --input "$WORK/absent.txt" \
+  --must-keep-file "$WORK/invalid.patterns" \
+  >"$WORK/invalid.out" 2>"$WORK/invalid.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 1 && ! -s "$WORK/invalid.out" ]] \
+  && grep -q "COMPRESS_FAIL invalid_must_keep_regex" "$WORK/invalid.err"; then
+  pass "invalid must-keep regex fails closed"
+else
+  fail "invalid regex did not fail closed (rc=$RC)"
+fi
+
+# 13) Non-empty raw input may not silently compress to an empty view.
+printf 'ordinary informational chatter\n' > "$WORK/empty-loss.txt"
+set +e
+python3 "$CC" --mode structural --input "$WORK/empty-loss.txt" \
+  >"$WORK/empty-loss.out" 2>"$WORK/empty-loss.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 1 ]] \
+  && cmp -s "$WORK/empty-loss.txt" "$WORK/empty-loss.out" \
+  && grep -q "COMPRESS_REJECTED fallback=raw" "$WORK/empty-loss.err"; then
+  pass "empty compressed view is rejected and raw is preserved"
+else
+  fail "non-empty raw input was allowed to disappear (rc=$RC)"
+fi
+
+# 14) Dedup may collapse repeated warnings only when the multiplicity marker
+# proves how many occurrences were present.
+printf '[WRN] repeated warning\n[WRN] repeated warning\n[WRN] repeated warning\n' > "$WORK/repeated-warning.txt"
+set +e
+python3 "$CC" --mode structural --input "$WORK/repeated-warning.txt" \
+  >"$WORK/repeated-warning.out" 2>"$WORK/repeated-warning.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 0 ]] \
+  && grep -q '\[WRN\] repeated warning  (x3)' "$WORK/repeated-warning.out"; then
+  pass "dedup preserves repeated-warning multiplicity"
+else
+  fail "repeated-warning multiplicity was lost (rc=$RC)"
+fi
+
+# 15) A literal line ending in "(xN)" is content, not necessarily a dedup
+# marker. It must survive in both non-dedup and dedup modes.
+printf 'ERROR literal marker  (x3)\n' > "$WORK/literal-marker.txt"
+set +e
+python3 "$CC" --mode extractive --input "$WORK/literal-marker.txt" \
+  >"$WORK/literal-marker.out" 2>"$WORK/literal-marker.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 0 ]] \
+  && cmp -s "$WORK/literal-marker.txt" "$WORK/literal-marker.out"; then
+  pass "extractive mode does not misread a literal (xN) suffix"
+else
+  fail "literal (xN) content triggered a false preservation failure (rc=$RC)"
+fi
+
+set +e
+python3 "$CC" --mode structural --input "$WORK/literal-marker.txt" \
+  >"$WORK/literal-marker-structural.out" 2>"$WORK/literal-marker-structural.err"
+RC=$?
+set -e 2>/dev/null || true
+if [[ "$RC" -eq 0 ]] \
+  && cmp -s "$WORK/literal-marker.txt" "$WORK/literal-marker-structural.out"; then
+  pass "structural mode disambiguates a literal (xN) suffix"
+else
+  fail "structural mode misread literal (xN) content (rc=$RC)"
+fi
+
 echo ""
 echo "===== Plan-08 ⟷ Control Plane: settings govern harness ====="
 CPS="$CASAN_HARNESS_ROOT/scripts/bash/control-plane-settings.py"