diff --git a/README.ko.md b/README.ko.md index 03afb09..ec69cf8 100644 --- a/README.ko.md +++ b/README.ko.md @@ -501,6 +501,12 @@ python3 scripts/release_smoke.py 버전별 릴리스 노트는 [CHANGELOG.md](CHANGELOG.md)에 기록하며, 사전 배포 게이트는 플러그인 매니페스트 버전과 일치하는 항목이 있는지 확인합니다. +### 실험적 semantic-GC plan gate + +`semantic-gc`는 기본 비활성화된 deny 전용 계획 검토 gate입니다. 기본 비활성화는 registry intent를 뜻하며, 명시적 plan CLI는 계속 실행할 수 있지만 omission이나 runtime action을 활성화하지 않습니다. 전체 envelope나 graph topology가 모호하면 graph evaluation을 억제합니다. 도달할 수 없는 node는 semantic irrelevance의 증명이 아니라 검토 후보일 뿐이며 omission과 runtime action은 승인되지 않습니다. missed-context note는 신뢰되지 않은 입력입니다. 이 planner는 context/artifact 내용을 읽지 않고 provenance, fallback, provider, hosted 절감을 검증하지 않습니다. Exit 0은 `ready_for_plan_review`만 뜻하며 delete/omit 권한이 아닙니다. + +context-guard experiments plan semantic-gc --json --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false}' --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"orphan","references":[],"is_root":false,"protected_zone":false,"content_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","provenance":{"source_label":"canonical-example","receipt_id":"0123456789abcdef"},"missed_context_note":"A reviewer could lose the orphaned rationale.","exact_fallback_command":"context-guard-artifact get 0123456789abcdef --full"}' --provider-boundary-ack --human-review-ack --protected-zone-policy deny + ## 라이선스 Copyright 2026 jinhongan. Apache License 2.0으로 배포됩니다. 자세한 내용은 [LICENSE](LICENSE)와 [NOTICE](NOTICE)를 참고하세요. diff --git a/README.md b/README.md index 9a2f0f7..af771a6 100644 --- a/README.md +++ b/README.md @@ -554,6 +554,12 @@ When a helper under `context-guard-kit/` changes, run `python3 scripts/sync_plug Versioned release notes live in [CHANGELOG.md](CHANGELOG.md); the prepublish gate requires an entry matching the plugin manifest version before publishing. +### Experimental semantic-GC plan gate + +`semantic-gc` is a default-off, deny-only, plan-review gate over a caller-declared graph. Default-off describes registry intent; the explicit plan CLI remains invocable and never enables omission or runtime action. Graph evaluation is suppressed when the complete envelope or topology is ambiguous. Unreachable nodes are review candidates, not proof of semantic irrelevance: omission and runtime action remain unauthorized. Candidate missed-context notes are untrusted. The planner does not read context/artifact content or verify provenance, fallback, providers, or hosted savings. Exit 0 means only `ready_for_plan_review`; it is never delete/omit authority. + +context-guard experiments plan semantic-gc --json --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false}' --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"orphan","references":[],"is_root":false,"protected_zone":false,"content_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","provenance":{"source_label":"canonical-example","receipt_id":"0123456789abcdef"},"missed_context_note":"A reviewer could lose the orphaned rationale.","exact_fallback_command":"context-guard-artifact get 0123456789abcdef --full"}' --provider-boundary-ack --human-review-ack --protected-zone-policy deny + ## License Copyright 2026 jinhongan. Licensed under the Apache License 2.0. See [LICENSE](LICENSE) and [NOTICE](NOTICE). diff --git a/context-guard-kit/README.md b/context-guard-kit/README.md index 5e6a6b0..c8c3099 100644 --- a/context-guard-kit/README.md +++ b/context-guard-kit/README.md @@ -114,6 +114,11 @@ context-guard experiments plan proof-carrying-context --json --proof-unit-json ' Claude Code에 적용하려면 `settings.example.json`을 `.claude/settings.json`으로 복사하되, 먼저 작은 repo에서 quoting/종료 코드를 확인하세요. +### Experimental semantic-GC plan gate + +`semantic-gc` is a default-off, deny-only, plan-review gate over a caller-declared graph. Default-off describes registry intent; the explicit plan CLI remains invocable and never enables omission or runtime action. Graph evaluation is suppressed when the complete envelope or topology is ambiguous. Unreachable nodes are review candidates, not proof of semantic irrelevance: omission and runtime action remain unauthorized. Candidate missed-context notes are untrusted. The planner does not read context/artifact content or verify provenance, fallback, providers, or hosted savings. Exit 0 means only `ready_for_plan_review`; it is never delete/omit authority. + +context-guard experiments plan semantic-gc --json --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false}' --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"orphan","references":[],"is_root":false,"protected_zone":false,"content_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","provenance":{"source_label":"canonical-example","receipt_id":"0123456789abcdef"},"missed_context_note":"A reviewer could lose the orphaned rationale.","exact_fallback_command":"context-guard-artifact get 0123456789abcdef --full"}' --provider-boundary-ack --human-review-ack --protected-zone-policy deny ## License diff --git a/context-guard-kit/experimental_registry.py b/context-guard-kit/experimental_registry.py index 92a7abc..20023ff 100755 --- a/context-guard-kit/experimental_registry.py +++ b/context-guard-kit/experimental_registry.py @@ -26,6 +26,7 @@ import socket from socketserver import TCPServer from pathlib import Path +import unicodedata import stat import sys import time @@ -75,6 +76,14 @@ PROOF_CARRYING_CONTEXT_DETAILED_UNIT_CAP = 64 PROOF_CARRYING_CONTEXT_UNIT_JSON_BYTE_CAP = 8192 PROOF_UNIT_JSON_MAX_DEPTH = 100 +SEMANTIC_GC_PLAN_SCHEMA_VERSION = "contextguard.experiments.semantic-gc-plan.v1" +SEMANTIC_GC_UNIT_SCHEMA_VERSION = "contextguard.semantic-gc-unit.v1" +SEMANTIC_GC_DETAILED_UNIT_CAP = 64 +SEMANTIC_GC_UNIT_JSON_BYTE_CAP = 8192 +SEMANTIC_GC_JSON_MAX_DEPTH = 100 +SEMANTIC_GC_PROCESS_EXIT_CONTRACT = ( + "exit code 0 means ready_for_plan_review; exit code 2 means a blocked plan was emitted" +) JSON_SAFE_INTEGER_MAX = 9_007_199_254_740_991 PROOF_SOURCE_LABEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+-]{0,119}$") PROOF_RECEIPT_ID_RE = re.compile(r"^[a-f0-9]{16,64}$") @@ -129,6 +138,36 @@ "safe_range_omitted", "duplicate_proof_unit", ) +SEMANTIC_GC_UNIT_ID_RE = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$") +SEMANTIC_GC_SOURCE_LABEL_RE = re.compile(r"^[A-Za-z0-9._:/ -]{1,128}$") +SEMANTIC_GC_RECEIPT_ID_RE = re.compile(r"^[a-f0-9]{16,64}$") +SEMANTIC_GC_CONTENT_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") +SEMANTIC_GC_ALLOWED_FIELDS = frozenset({ + "schema", "unit_id", "references", "is_root", "protected_zone", + "content_sha256", "provenance", "missed_context_note", "exact_fallback_command", +}) +SEMANTIC_GC_BLOCKER_ORDER = ( + "no_context_units", "unit_limit_exceeded", "invalid_context_unit_json", + "duplicate_json_key", "context_unit_depth_exceeded", "nonfinite_json_number", + "invalid_unicode_scalar", "decoder_recursion_limit", "invalid_context_unit_schema", + "unknown_context_unit_field", "missing_unit_id", "invalid_unit_id", + "duplicate_unit_id", "invalid_references", "duplicate_reference", "unknown_reference", + "ambiguous_reference", + "invalid_root_flag", "invalid_protected_zone_flag", "no_declared_root", + "graph_evaluation_suppressed", "protected_zone_policy_required", + "invalid_content_sha256", "missing_provenance", "invalid_provenance", + "invalid_source_label", "invalid_receipt_id", "missing_missed_context_note", + "invalid_missed_context_note", "missing_exact_fallback", "invalid_exact_fallback", + "fallback_receipt_mismatch", "provider_boundary_ack_required", "human_review_ack_required", +) +SEMANTIC_GC_WARNING_ORDER = ( + "plan_only_no_omission", "caller_declared_graph_unverified", + "semantic_relevance_not_evaluated", "provider_boundary_not_verified", + "provenance_not_verified_externally", "fallback_not_executed", + "human_review_still_required", "accepted_notes_are_untrusted", + "duplicate_content_sha256", "duplicate_receipt_id", + "protected_unreachable_excluded", "no_sweep_candidates", +) IMAGE_CONTEXT_PACK_PROVIDER_BOUNDARY = "provider-measured-matched-tasks-required" LOCAL_PROXY_DEFAULT_BIND_HOST = "127.0.0.1" LOCAL_PROXY_DEFAULT_BIND_PORT = 0 @@ -458,6 +497,46 @@ def to_json(self, *, enabled: bool = False) -> dict[str, Any]: "and rehydration remain explicitly unchecked." ), ), + Experiment( + id="semantic-gc", + name="Semantic graph garbage-collection planning gate", + summary=( + "Plan-only caller-declared mark-and-sweep classification with strict graph-integrity suppression " + "and recovery-evidence gates for human review." + ), + stability="experimental", + default_enabled=False, + risk_level="high", + claim_boundary=( + "Unreachable graph nodes are plan-review candidates, not proof of irrelevance or authorization to omit; " + "the planner does not read content, verify provenance, execute fallback, or call providers." + ), + gate_requirements=( + "complete unambiguous caller-declared graph", + "deny-only protected-zone declaration", + "candidate recovery evidence and missed-context note", + "provider-boundary acknowledgement for every complete graph; " + "human-review acknowledgement when unprotected sweep candidates exist", + ), + runtime_status="available-plan-only", + commands=("context-guard experiments plan semantic-gc",), + opt_in_flags=( + "plan semantic-gc", "--context-unit-json", "--provider-boundary-ack", + "--human-review-ack", "--protected-zone-policy deny", + ), + config_effect=( + "Registry enablement records project-local intent only; semantic-gc exposes one deterministic plan command. " + "It does not add an emit/record/serve/apply/delete/omit runtime, read context or artifacts, write files, " + "call models/providers/network, execute fallback, replace context, or authorize omission." + ), + evidence_contract=( + "The complete caller-declared graph must pass strict structural validation before iterative reachability. " + "Unprotected unreachable candidates require sanitized provenance, content hash, exact fallback, and an " + "untrusted missed-context note plus human-review acknowledgement before the plan is ready for review. " + "A complete graph with no unprotected sweep candidates does not require that acknowledgement. " + "Ready plans exit 0; blocked plans still emit their envelope and exit 2." + ), + ), Experiment( id="learned-compression", name="Learned/synthetic compression candidate gate", @@ -2855,6 +2934,425 @@ def command_plan_proof_carrying_context(args: argparse.Namespace) -> int: return 0 +_SEMANTIC_GC_NONFINITE_SENTINEL = object() + + +def ordered_semantic_gc_taxonomy(values: list[str] | set[str], order: tuple[str, ...]) -> list[str]: + selected = set(values) + return [value for value in order if value in selected] + + +def semantic_gc_validation_row(index: int, issue: str | None = None) -> dict[str, Any]: + return { + "candidate_safety_applicable": None, + "candidate_safety_issues": [], + "input_index": index, + "structural_issues": [issue] if issue else [], + "unit_id": None, + } + + +def decode_semantic_gc_unit(raw: Any, index: int) -> tuple[dict[str, Any] | None, dict[str, Any]]: + row = semantic_gc_validation_row(index) + if not isinstance(raw, str): + row["structural_issues"] = ["invalid_context_unit_json"] + return None, row + try: + encoded = raw.encode("utf-8", errors="strict") + except UnicodeEncodeError: + row["structural_issues"] = ["invalid_unicode_scalar"] + return None, row + if len(encoded) > SEMANTIC_GC_UNIT_JSON_BYTE_CAP: + row["structural_issues"] = ["invalid_context_unit_json"] + return None, row + + duplicate_key = False + + def pairs_hook(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + nonlocal duplicate_key + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + duplicate_key = True + result[key] = value + return result + + try: + decoded = json.loads( + raw, + object_pairs_hook=pairs_hook, + parse_constant=lambda _value: _SEMANTIC_GC_NONFINITE_SENTINEL, + ) + except RecursionError: + row["structural_issues"] = ["decoder_recursion_limit"] + return None, row + except (json.JSONDecodeError, UnicodeDecodeError, ValueError, TypeError): + row["structural_issues"] = ["invalid_context_unit_json"] + return None, row + + depth_exceeded = False + nonfinite = False + invalid_unicode = False + stack: list[tuple[Any, int]] = [(decoded, 0)] + while stack: + value, depth = stack.pop() + if depth > SEMANTIC_GC_JSON_MAX_DEPTH: + depth_exceeded = True + continue + if isinstance(value, str): + try: + value.encode("utf-8", errors="strict") + except UnicodeEncodeError: + invalid_unicode = True + elif value is _SEMANTIC_GC_NONFINITE_SENTINEL or (type(value) is float and not math.isfinite(value)): + nonfinite = True + if isinstance(value, dict): + for key, child in value.items(): + stack.append((key, depth + 1)) + stack.append((child, depth + 1)) + elif isinstance(value, list): + for child in value: + stack.append((child, depth + 1)) + + issue = None + if duplicate_key: + issue = "duplicate_json_key" + elif depth_exceeded: + issue = "context_unit_depth_exceeded" + elif nonfinite: + issue = "nonfinite_json_number" + elif invalid_unicode: + issue = "invalid_unicode_scalar" + elif not isinstance(decoded, dict): + issue = "invalid_context_unit_json" + if issue: + row["structural_issues"] = [issue] + return None, row + return decoded, row + + +def normalize_semantic_gc_structure(obj: dict[str, Any], row: dict[str, Any]) -> dict[str, Any]: + issues: list[str] = [] + if obj.get("schema") != SEMANTIC_GC_UNIT_SCHEMA_VERSION: + issues.append("invalid_context_unit_schema") + if set(obj) - SEMANTIC_GC_ALLOWED_FIELDS: + issues.append("unknown_context_unit_field") + + raw_id = obj.get("unit_id") + unit_id = raw_id if isinstance(raw_id, str) and SEMANTIC_GC_UNIT_ID_RE.fullmatch(raw_id) else None + if raw_id is None or raw_id == "": + issues.append("missing_unit_id") + elif unit_id is None: + issues.append("invalid_unit_id") + row["unit_id"] = unit_id + + raw_references = obj.get("references") + references: list[str] = [] + reference_targets: list[str] = [] + if not isinstance(raw_references, list) or len(raw_references) > 64: + issues.append("invalid_references") + else: + seen: set[str] = set() + reference_issue = False + for reference in raw_references: + if not isinstance(reference, str) or SEMANTIC_GC_UNIT_ID_RE.fullmatch(reference) is None: + issues.append("invalid_references") + reference_issue = True + continue + if reference in seen: + issues.append("duplicate_reference") + reference_issue = True + continue + references.append(reference) + reference_targets.append(reference) + seen.add(reference) + if reference_issue: + references = [] + + is_root = obj.get("is_root") + if type(is_root) is not bool: + issues.append("invalid_root_flag") + is_root = None + protected = obj.get("protected_zone") + if type(protected) is not bool: + issues.append("invalid_protected_zone_flag") + protected = None + row["structural_issues"] = ordered_semantic_gc_taxonomy(issues, SEMANTIC_GC_BLOCKER_ORDER) + return { + "object": obj, + "row": row, + "unit_id": unit_id, + "references": references, + "reference_targets": reference_targets, + "is_root": is_root, + "protected_zone": protected, + } + + +def valid_semantic_gc_note(value: Any) -> bool: + if not isinstance(value, str) or not (1 <= len(value) <= 512) or value != value.strip(): + return False + return not any(unicodedata.category(char) in {"Cc", "Cf", "Zl", "Zp"} for char in value) + + +def normalize_semantic_gc_candidate(unit: dict[str, Any]) -> tuple[dict[str, Any], list[str], str | None, str | None]: + obj = unit["object"] + issues: list[str] = [] + raw_hash = obj.get("content_sha256") + content_hash = raw_hash if isinstance(raw_hash, str) and SEMANTIC_GC_CONTENT_SHA256_RE.fullmatch(raw_hash) else None + if content_hash is None: + issues.append("invalid_content_sha256") + + raw_provenance = obj.get("provenance") + source_label = None + receipt_id = None + if raw_provenance is None: + issues.append("missing_provenance") + elif not isinstance(raw_provenance, dict) or set(raw_provenance) != {"source_label", "receipt_id"}: + issues.append("invalid_provenance") + else: + raw_label = raw_provenance.get("source_label") + if ( + isinstance(raw_label, str) + and raw_label == raw_label.strip() + and SEMANTIC_GC_SOURCE_LABEL_RE.fullmatch(raw_label) is not None + ): + source_label = raw_label + else: + issues.append("invalid_source_label") + raw_receipt = raw_provenance.get("receipt_id") + if isinstance(raw_receipt, str) and SEMANTIC_GC_RECEIPT_ID_RE.fullmatch(raw_receipt): + receipt_id = raw_receipt + else: + issues.append("invalid_receipt_id") + + raw_note = obj.get("missed_context_note") + note = raw_note if valid_semantic_gc_note(raw_note) else None + if raw_note is None or raw_note == "": + issues.append("missing_missed_context_note") + elif note is None: + issues.append("invalid_missed_context_note") + + raw_fallback = obj.get("exact_fallback_command") + fallback = None + if raw_fallback is None or raw_fallback == "": + issues.append("missing_exact_fallback") + elif not isinstance(raw_fallback, str): + issues.append("invalid_exact_fallback") + else: + forbidden = (';', '|', '&', '>', '<', '`', '$', '\\', '\n', '\r', '"', "'") + if any(token in raw_fallback for token in forbidden) or raw_fallback != raw_fallback.strip(): + issues.append("invalid_exact_fallback") + else: + parts = raw_fallback.split(" ") + if len(parts) != 4 or parts[0] != "context-guard-artifact" or parts[1] != "get" or parts[3] != "--full": + issues.append("invalid_exact_fallback") + elif SEMANTIC_GC_RECEIPT_ID_RE.fullmatch(parts[2]) is None: + issues.append("invalid_exact_fallback") + elif receipt_id is not None and parts[2] != receipt_id: + issues.append("fallback_receipt_mismatch") + elif receipt_id is None: + issues.append("invalid_exact_fallback") + else: + fallback = raw_fallback + + issues = ordered_semantic_gc_taxonomy(issues, SEMANTIC_GC_BLOCKER_ORDER) + return ({ + "candidate_replacement": None, + "candidate_safety_issues": issues, + "content_sha256": content_hash, + "exact_fallback_command": fallback, + "human_review_required": True, + "missed_context_note": note, + "protected_zone": False, + "provenance": {"receipt_id": receipt_id, "source_label": source_label}, + "reason": "unreachable_from_declared_roots", + "unit_id": unit["unit_id"], + }, issues, content_hash, receipt_id) + + +def semantic_gc_plan_payload(args: argparse.Namespace) -> dict[str, Any]: + raw_units = args.context_unit_json or [] + total_count = len(raw_units) + detailed_count = min(total_count, SEMANTIC_GC_DETAILED_UNIT_CAP) + overflow_count = max(total_count - SEMANTIC_GC_DETAILED_UNIT_CAP, 0) + units: list[dict[str, Any]] = [] + rows: list[dict[str, Any]] = [] + decoded_count = 0 + for index, raw in enumerate(raw_units[:SEMANTIC_GC_DETAILED_UNIT_CAP]): + decoded, row = decode_semantic_gc_unit(raw, index) + rows.append(row) + if decoded is None: + continue + decoded_count += 1 + units.append(normalize_semantic_gc_structure(decoded, row)) + + by_id: dict[str, list[dict[str, Any]]] = {} + for unit in units: + if unit["unit_id"] is not None: + by_id.setdefault(unit["unit_id"], []).append(unit) + duplicate_ids = {unit_id for unit_id, matches in by_id.items() if len(matches) > 1} + for unit in units: + structural = list(unit["row"]["structural_issues"]) + if unit["unit_id"] in duplicate_ids: + structural.append("duplicate_unit_id") + if any(reference not in by_id for reference in unit["reference_targets"]): + structural.append("unknown_reference") + if any(reference in duplicate_ids for reference in unit["reference_targets"]): + structural.append("ambiguous_reference") + unit["row"]["structural_issues"] = ordered_semantic_gc_taxonomy(structural, SEMANTIC_GC_BLOCKER_ORDER) + + declared_roots = sorted( + unit_id for unit_id, matches in by_id.items() + if unit_id not in duplicate_ids and len(matches) == 1 and matches[0]["is_root"] is True + ) + structural_blockers: list[str] = [] + if total_count == 0: + structural_blockers.append("no_context_units") + if overflow_count: + structural_blockers.append("unit_limit_exceeded") + for row in rows: + structural_blockers.extend(row["structural_issues"]) + if not declared_roots: + structural_blockers.append("no_declared_root") + graph_complete = not structural_blockers + + marked_ids: list[str] = [] + candidates: list[dict[str, Any]] = [] + unreachable_count = 0 + protected_unreachable_count = 0 + safety_valid_count = 0 + safety_invalid_count = 0 + safety_blockers: list[str] = [] + candidate_hashes: list[str] = [] + candidate_receipts: list[str] = [] + if graph_complete: + marked: set[str] = set() + pending = list(declared_roots) + while pending: + unit_id = pending.pop() + if unit_id in marked: + continue + marked.add(unit_id) + pending.extend(by_id[unit_id][0]["references"]) + marked_ids = sorted(marked) + for unit in units: + unit_id = unit["unit_id"] + assert unit_id is not None + row = unit["row"] + if unit_id in marked: + row["candidate_safety_applicable"] = False + continue + unreachable_count += 1 + if unit["protected_zone"] is True: + protected_unreachable_count += 1 + row["candidate_safety_applicable"] = False + continue + row["candidate_safety_applicable"] = True + candidate, issues, content_hash, receipt_id = normalize_semantic_gc_candidate(unit) + row["candidate_safety_issues"] = issues + candidates.append(candidate) + safety_blockers.extend(issues) + if issues: + safety_invalid_count += 1 + else: + safety_valid_count += 1 + if content_hash is not None: + candidate_hashes.append(content_hash) + if receipt_id is not None: + candidate_receipts.append(receipt_id) + candidates.sort(key=lambda item: item["unit_id"]) + + blockers = list(structural_blockers) + if not graph_complete: + blockers.append("graph_evaluation_suppressed") + protected_policy = "deny" if getattr(args, "protected_zone_policy", None) == "deny" else None + if protected_policy is None: + blockers.append("protected_zone_policy_required") + blockers.extend(safety_blockers) + if graph_complete and not args.provider_boundary_ack: + blockers.append("provider_boundary_ack_required") + if graph_complete and candidates and not args.human_review_ack: + blockers.append("human_review_ack_required") + blockers = ordered_semantic_gc_taxonomy(blockers, SEMANTIC_GC_BLOCKER_ORDER) + + warnings = list(SEMANTIC_GC_WARNING_ORDER[:6]) + if candidates: + warnings.extend(("human_review_still_required", "accepted_notes_are_untrusted")) + if len(candidate_hashes) != len(set(candidate_hashes)): + warnings.append("duplicate_content_sha256") + if len(candidate_receipts) != len(set(candidate_receipts)): + warnings.append("duplicate_receipt_id") + if protected_unreachable_count: + warnings.append("protected_unreachable_excluded") + if graph_complete and not candidates: + warnings.append("no_sweep_candidates") + warnings = ordered_semantic_gc_taxonomy(warnings, SEMANTIC_GC_WARNING_ORDER) + + verification_scope = { + "artifact_content_read": False, + "context_content_read": False, + "deletion_or_omission_performed": False, + "exact_fallback_executed": False, + "files_written": False, + "model_or_provider_called": False, + "network_called": False, + "provenance_verified_externally": False, + "subprocess_started": False, + } + return { + "blockers": blockers, + "candidate_count": len(candidates), + "candidate_replacement": None, + "candidate_safety_invalid_count": safety_invalid_count, + "candidate_safety_valid_count": safety_valid_count, + "candidates": candidates, + "declared_root_count": len(declared_roots), + "declared_root_ids": declared_roots, + "decoded_unit_count": decoded_count, + "detailed_unit_count": detailed_count, + "effective_protected_zone_policy": "deny", + "experiment": "semantic-gc", + "graph_evaluation_performed": graph_complete, + "graph_integrity_complete": graph_complete, + "human_review_acknowledged": bool(args.human_review_ack), + "human_review_performed": False, + "marked_unit_count": len(marked_ids), + "marked_unit_ids": marked_ids, + "omission_authorized": False, + "overflow_unit_count": overflow_count, + "plan_only": True, + "process_exit_contract": SEMANTIC_GC_PROCESS_EXIT_CONTRACT, + "protected_unreachable_count": protected_unreachable_count, + "protected_zone_policy": protected_policy, + "provider_boundary_acknowledged": bool(args.provider_boundary_ack), + "runtime_action_allowed": False, + "schema": SEMANTIC_GC_PLAN_SCHEMA_VERSION, + "status": "ready_for_plan_review" if not blockers else "blocked", + "structurally_valid_unit_count": sum(not row["structural_issues"] for row in rows), + "total_unit_count": total_count, + "unit_validation": rows, + "unreachable_unit_count": unreachable_count, + "verification_scope": verification_scope, + "warnings": warnings, + } + + +def command_plan_semantic_gc(args: argparse.Namespace) -> int: + payload = semantic_gc_plan_payload(args) + if args.json: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) + else: + print("ContextGuard semantic-gc plan") + print(f"Status: {payload['status']}") + print(f"Graph complete: {payload['graph_integrity_complete']}; candidates: {payload['candidate_count']}") + if payload["blockers"]: + print(f"Blockers: {', '.join(payload['blockers'])}") + print(f"Exit contract: {payload['process_exit_contract']}.") + print("semantic-gc is plan-only; no context was deleted, omitted, read, replaced, or authorized for runtime action.") + return 0 if payload["status"] == "ready_for_plan_review" else 2 + + def visual_crop_ocr_evidence_pack_payload(args: argparse.Namespace) -> dict[str, Any]: payload = visual_crop_ocr_plan_payload(args) blockers = list(payload["review_plan"]["readiness_blockers"]) @@ -5662,6 +6160,35 @@ def build_parser() -> argparse.ArgumentParser: proof_carrying_context.add_argument("--json", action="store_true", help="Emit JSON output.") proof_carrying_context.set_defaults(func=command_plan_proof_carrying_context) + semantic_gc = plan_sub.add_parser( + "semantic-gc", + allow_abbrev=False, + help="Plan caller-declared graph reachability candidates without reading or omitting context.", + ) + semantic_gc.add_argument( + "--context-unit-json", + action="append", + help="Inline literal semantic-GC unit JSON object. Repeatable; never treated as a path.", + ) + semantic_gc.add_argument( + "--provider-boundary-ack", + action="store_true", + help="Acknowledge that provider behavior and hosted savings remain unverified.", + ) + semantic_gc.add_argument( + "--human-review-ack", + action="store_true", + help="Acknowledge that candidate review remains required; this does not perform review.", + ) + semantic_gc.add_argument( + "--protected-zone-policy", + choices=("deny",), + default=None, + help="Explicit deny-only protected-zone declaration; omitted remains effective deny but blocks readiness.", + ) + semantic_gc.add_argument("--json", action="store_true", help="Emit JSON output.") + semantic_gc.set_defaults(func=command_plan_semantic_gc) + self_hosted = plan_sub.add_parser( "self-hosted-metrics-ledger", help="Dry-run self-hosted/local metrics ledger sidecar evidence without writing a ledger.", diff --git a/plugins/context-guard/README.ko.md b/plugins/context-guard/README.ko.md index 97b875e..109d6b2 100644 --- a/plugins/context-guard/README.ko.md +++ b/plugins/context-guard/README.ko.md @@ -151,6 +151,12 @@ claude --plugin-dir ./plugins/context-guard /plugin install context-guard@context-guard ``` +### 실험적 semantic-GC plan gate + +`semantic-gc`는 기본 비활성화된 deny 전용 계획 검토 gate입니다. 기본 비활성화는 registry intent를 뜻하며, 명시적 plan CLI는 계속 실행할 수 있지만 omission이나 runtime action을 활성화하지 않습니다. 전체 envelope나 graph topology가 모호하면 graph evaluation을 억제합니다. 도달할 수 없는 node는 semantic irrelevance의 증명이 아니라 검토 후보일 뿐이며 omission과 runtime action은 승인되지 않습니다. missed-context note는 신뢰되지 않은 입력입니다. 이 planner는 context/artifact 내용을 읽지 않고 provenance, fallback, provider, hosted 절감을 검증하지 않습니다. Exit 0은 `ready_for_plan_review`만 뜻하며 delete/omit 권한이 아닙니다. + +context-guard experiments plan semantic-gc --json --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false}' --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"orphan","references":[],"is_root":false,"protected_zone":false,"content_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","provenance":{"source_label":"canonical-example","receipt_id":"0123456789abcdef"},"missed_context_note":"A reviewer could lose the orphaned rationale.","exact_fallback_command":"context-guard-artifact get 0123456789abcdef --full"}' --provider-boundary-ack --human-review-ack --protected-zone-policy deny + ## 라이선스 Copyright 2026 jinhongan. Apache License 2.0으로 배포됩니다. [LICENSE](LICENSE)와 [NOTICE](NOTICE)를 참고하세요. diff --git a/plugins/context-guard/README.md b/plugins/context-guard/README.md index 3c15822..32e8428 100644 --- a/plugins/context-guard/README.md +++ b/plugins/context-guard/README.md @@ -187,6 +187,12 @@ Marketplace installation test: /plugin install context-guard@context-guard ``` +### Experimental semantic-GC plan gate + +`semantic-gc` is a default-off, deny-only, plan-review gate over a caller-declared graph. Default-off describes registry intent; the explicit plan CLI remains invocable and never enables omission or runtime action. Graph evaluation is suppressed when the complete envelope or topology is ambiguous. Unreachable nodes are review candidates, not proof of semantic irrelevance: omission and runtime action remain unauthorized. Candidate missed-context notes are untrusted. The planner does not read context/artifact content or verify provenance, fallback, providers, or hosted savings. Exit 0 means only `ready_for_plan_review`; it is never delete/omit authority. + +context-guard experiments plan semantic-gc --json --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false}' --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"orphan","references":[],"is_root":false,"protected_zone":false,"content_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","provenance":{"source_label":"canonical-example","receipt_id":"0123456789abcdef"},"missed_context_note":"A reviewer could lose the orphaned rationale.","exact_fallback_command":"context-guard-artifact get 0123456789abcdef --full"}' --provider-boundary-ack --human-review-ack --protected-zone-policy deny + ## License Copyright 2026 jinhongan. Licensed under the Apache License 2.0. See [LICENSE](LICENSE) and [NOTICE](NOTICE). diff --git a/plugins/context-guard/bin/context-guard-experiments b/plugins/context-guard/bin/context-guard-experiments index 92a7abc..20023ff 100755 --- a/plugins/context-guard/bin/context-guard-experiments +++ b/plugins/context-guard/bin/context-guard-experiments @@ -26,6 +26,7 @@ import shlex import socket from socketserver import TCPServer from pathlib import Path +import unicodedata import stat import sys import time @@ -75,6 +76,14 @@ PROOF_CARRYING_CONTEXT_UNIT_SCHEMA_VERSION = "contextguard.proof-unit.v1" PROOF_CARRYING_CONTEXT_DETAILED_UNIT_CAP = 64 PROOF_CARRYING_CONTEXT_UNIT_JSON_BYTE_CAP = 8192 PROOF_UNIT_JSON_MAX_DEPTH = 100 +SEMANTIC_GC_PLAN_SCHEMA_VERSION = "contextguard.experiments.semantic-gc-plan.v1" +SEMANTIC_GC_UNIT_SCHEMA_VERSION = "contextguard.semantic-gc-unit.v1" +SEMANTIC_GC_DETAILED_UNIT_CAP = 64 +SEMANTIC_GC_UNIT_JSON_BYTE_CAP = 8192 +SEMANTIC_GC_JSON_MAX_DEPTH = 100 +SEMANTIC_GC_PROCESS_EXIT_CONTRACT = ( + "exit code 0 means ready_for_plan_review; exit code 2 means a blocked plan was emitted" +) JSON_SAFE_INTEGER_MAX = 9_007_199_254_740_991 PROOF_SOURCE_LABEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+-]{0,119}$") PROOF_RECEIPT_ID_RE = re.compile(r"^[a-f0-9]{16,64}$") @@ -129,6 +138,36 @@ PROOF_WARNING_ORDER = ( "safe_range_omitted", "duplicate_proof_unit", ) +SEMANTIC_GC_UNIT_ID_RE = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$") +SEMANTIC_GC_SOURCE_LABEL_RE = re.compile(r"^[A-Za-z0-9._:/ -]{1,128}$") +SEMANTIC_GC_RECEIPT_ID_RE = re.compile(r"^[a-f0-9]{16,64}$") +SEMANTIC_GC_CONTENT_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") +SEMANTIC_GC_ALLOWED_FIELDS = frozenset({ + "schema", "unit_id", "references", "is_root", "protected_zone", + "content_sha256", "provenance", "missed_context_note", "exact_fallback_command", +}) +SEMANTIC_GC_BLOCKER_ORDER = ( + "no_context_units", "unit_limit_exceeded", "invalid_context_unit_json", + "duplicate_json_key", "context_unit_depth_exceeded", "nonfinite_json_number", + "invalid_unicode_scalar", "decoder_recursion_limit", "invalid_context_unit_schema", + "unknown_context_unit_field", "missing_unit_id", "invalid_unit_id", + "duplicate_unit_id", "invalid_references", "duplicate_reference", "unknown_reference", + "ambiguous_reference", + "invalid_root_flag", "invalid_protected_zone_flag", "no_declared_root", + "graph_evaluation_suppressed", "protected_zone_policy_required", + "invalid_content_sha256", "missing_provenance", "invalid_provenance", + "invalid_source_label", "invalid_receipt_id", "missing_missed_context_note", + "invalid_missed_context_note", "missing_exact_fallback", "invalid_exact_fallback", + "fallback_receipt_mismatch", "provider_boundary_ack_required", "human_review_ack_required", +) +SEMANTIC_GC_WARNING_ORDER = ( + "plan_only_no_omission", "caller_declared_graph_unverified", + "semantic_relevance_not_evaluated", "provider_boundary_not_verified", + "provenance_not_verified_externally", "fallback_not_executed", + "human_review_still_required", "accepted_notes_are_untrusted", + "duplicate_content_sha256", "duplicate_receipt_id", + "protected_unreachable_excluded", "no_sweep_candidates", +) IMAGE_CONTEXT_PACK_PROVIDER_BOUNDARY = "provider-measured-matched-tasks-required" LOCAL_PROXY_DEFAULT_BIND_HOST = "127.0.0.1" LOCAL_PROXY_DEFAULT_BIND_PORT = 0 @@ -458,6 +497,46 @@ EXPERIMENTS: tuple[Experiment, ...] = ( "and rehydration remain explicitly unchecked." ), ), + Experiment( + id="semantic-gc", + name="Semantic graph garbage-collection planning gate", + summary=( + "Plan-only caller-declared mark-and-sweep classification with strict graph-integrity suppression " + "and recovery-evidence gates for human review." + ), + stability="experimental", + default_enabled=False, + risk_level="high", + claim_boundary=( + "Unreachable graph nodes are plan-review candidates, not proof of irrelevance or authorization to omit; " + "the planner does not read content, verify provenance, execute fallback, or call providers." + ), + gate_requirements=( + "complete unambiguous caller-declared graph", + "deny-only protected-zone declaration", + "candidate recovery evidence and missed-context note", + "provider-boundary acknowledgement for every complete graph; " + "human-review acknowledgement when unprotected sweep candidates exist", + ), + runtime_status="available-plan-only", + commands=("context-guard experiments plan semantic-gc",), + opt_in_flags=( + "plan semantic-gc", "--context-unit-json", "--provider-boundary-ack", + "--human-review-ack", "--protected-zone-policy deny", + ), + config_effect=( + "Registry enablement records project-local intent only; semantic-gc exposes one deterministic plan command. " + "It does not add an emit/record/serve/apply/delete/omit runtime, read context or artifacts, write files, " + "call models/providers/network, execute fallback, replace context, or authorize omission." + ), + evidence_contract=( + "The complete caller-declared graph must pass strict structural validation before iterative reachability. " + "Unprotected unreachable candidates require sanitized provenance, content hash, exact fallback, and an " + "untrusted missed-context note plus human-review acknowledgement before the plan is ready for review. " + "A complete graph with no unprotected sweep candidates does not require that acknowledgement. " + "Ready plans exit 0; blocked plans still emit their envelope and exit 2." + ), + ), Experiment( id="learned-compression", name="Learned/synthetic compression candidate gate", @@ -2855,6 +2934,425 @@ def command_plan_proof_carrying_context(args: argparse.Namespace) -> int: return 0 +_SEMANTIC_GC_NONFINITE_SENTINEL = object() + + +def ordered_semantic_gc_taxonomy(values: list[str] | set[str], order: tuple[str, ...]) -> list[str]: + selected = set(values) + return [value for value in order if value in selected] + + +def semantic_gc_validation_row(index: int, issue: str | None = None) -> dict[str, Any]: + return { + "candidate_safety_applicable": None, + "candidate_safety_issues": [], + "input_index": index, + "structural_issues": [issue] if issue else [], + "unit_id": None, + } + + +def decode_semantic_gc_unit(raw: Any, index: int) -> tuple[dict[str, Any] | None, dict[str, Any]]: + row = semantic_gc_validation_row(index) + if not isinstance(raw, str): + row["structural_issues"] = ["invalid_context_unit_json"] + return None, row + try: + encoded = raw.encode("utf-8", errors="strict") + except UnicodeEncodeError: + row["structural_issues"] = ["invalid_unicode_scalar"] + return None, row + if len(encoded) > SEMANTIC_GC_UNIT_JSON_BYTE_CAP: + row["structural_issues"] = ["invalid_context_unit_json"] + return None, row + + duplicate_key = False + + def pairs_hook(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + nonlocal duplicate_key + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + duplicate_key = True + result[key] = value + return result + + try: + decoded = json.loads( + raw, + object_pairs_hook=pairs_hook, + parse_constant=lambda _value: _SEMANTIC_GC_NONFINITE_SENTINEL, + ) + except RecursionError: + row["structural_issues"] = ["decoder_recursion_limit"] + return None, row + except (json.JSONDecodeError, UnicodeDecodeError, ValueError, TypeError): + row["structural_issues"] = ["invalid_context_unit_json"] + return None, row + + depth_exceeded = False + nonfinite = False + invalid_unicode = False + stack: list[tuple[Any, int]] = [(decoded, 0)] + while stack: + value, depth = stack.pop() + if depth > SEMANTIC_GC_JSON_MAX_DEPTH: + depth_exceeded = True + continue + if isinstance(value, str): + try: + value.encode("utf-8", errors="strict") + except UnicodeEncodeError: + invalid_unicode = True + elif value is _SEMANTIC_GC_NONFINITE_SENTINEL or (type(value) is float and not math.isfinite(value)): + nonfinite = True + if isinstance(value, dict): + for key, child in value.items(): + stack.append((key, depth + 1)) + stack.append((child, depth + 1)) + elif isinstance(value, list): + for child in value: + stack.append((child, depth + 1)) + + issue = None + if duplicate_key: + issue = "duplicate_json_key" + elif depth_exceeded: + issue = "context_unit_depth_exceeded" + elif nonfinite: + issue = "nonfinite_json_number" + elif invalid_unicode: + issue = "invalid_unicode_scalar" + elif not isinstance(decoded, dict): + issue = "invalid_context_unit_json" + if issue: + row["structural_issues"] = [issue] + return None, row + return decoded, row + + +def normalize_semantic_gc_structure(obj: dict[str, Any], row: dict[str, Any]) -> dict[str, Any]: + issues: list[str] = [] + if obj.get("schema") != SEMANTIC_GC_UNIT_SCHEMA_VERSION: + issues.append("invalid_context_unit_schema") + if set(obj) - SEMANTIC_GC_ALLOWED_FIELDS: + issues.append("unknown_context_unit_field") + + raw_id = obj.get("unit_id") + unit_id = raw_id if isinstance(raw_id, str) and SEMANTIC_GC_UNIT_ID_RE.fullmatch(raw_id) else None + if raw_id is None or raw_id == "": + issues.append("missing_unit_id") + elif unit_id is None: + issues.append("invalid_unit_id") + row["unit_id"] = unit_id + + raw_references = obj.get("references") + references: list[str] = [] + reference_targets: list[str] = [] + if not isinstance(raw_references, list) or len(raw_references) > 64: + issues.append("invalid_references") + else: + seen: set[str] = set() + reference_issue = False + for reference in raw_references: + if not isinstance(reference, str) or SEMANTIC_GC_UNIT_ID_RE.fullmatch(reference) is None: + issues.append("invalid_references") + reference_issue = True + continue + if reference in seen: + issues.append("duplicate_reference") + reference_issue = True + continue + references.append(reference) + reference_targets.append(reference) + seen.add(reference) + if reference_issue: + references = [] + + is_root = obj.get("is_root") + if type(is_root) is not bool: + issues.append("invalid_root_flag") + is_root = None + protected = obj.get("protected_zone") + if type(protected) is not bool: + issues.append("invalid_protected_zone_flag") + protected = None + row["structural_issues"] = ordered_semantic_gc_taxonomy(issues, SEMANTIC_GC_BLOCKER_ORDER) + return { + "object": obj, + "row": row, + "unit_id": unit_id, + "references": references, + "reference_targets": reference_targets, + "is_root": is_root, + "protected_zone": protected, + } + + +def valid_semantic_gc_note(value: Any) -> bool: + if not isinstance(value, str) or not (1 <= len(value) <= 512) or value != value.strip(): + return False + return not any(unicodedata.category(char) in {"Cc", "Cf", "Zl", "Zp"} for char in value) + + +def normalize_semantic_gc_candidate(unit: dict[str, Any]) -> tuple[dict[str, Any], list[str], str | None, str | None]: + obj = unit["object"] + issues: list[str] = [] + raw_hash = obj.get("content_sha256") + content_hash = raw_hash if isinstance(raw_hash, str) and SEMANTIC_GC_CONTENT_SHA256_RE.fullmatch(raw_hash) else None + if content_hash is None: + issues.append("invalid_content_sha256") + + raw_provenance = obj.get("provenance") + source_label = None + receipt_id = None + if raw_provenance is None: + issues.append("missing_provenance") + elif not isinstance(raw_provenance, dict) or set(raw_provenance) != {"source_label", "receipt_id"}: + issues.append("invalid_provenance") + else: + raw_label = raw_provenance.get("source_label") + if ( + isinstance(raw_label, str) + and raw_label == raw_label.strip() + and SEMANTIC_GC_SOURCE_LABEL_RE.fullmatch(raw_label) is not None + ): + source_label = raw_label + else: + issues.append("invalid_source_label") + raw_receipt = raw_provenance.get("receipt_id") + if isinstance(raw_receipt, str) and SEMANTIC_GC_RECEIPT_ID_RE.fullmatch(raw_receipt): + receipt_id = raw_receipt + else: + issues.append("invalid_receipt_id") + + raw_note = obj.get("missed_context_note") + note = raw_note if valid_semantic_gc_note(raw_note) else None + if raw_note is None or raw_note == "": + issues.append("missing_missed_context_note") + elif note is None: + issues.append("invalid_missed_context_note") + + raw_fallback = obj.get("exact_fallback_command") + fallback = None + if raw_fallback is None or raw_fallback == "": + issues.append("missing_exact_fallback") + elif not isinstance(raw_fallback, str): + issues.append("invalid_exact_fallback") + else: + forbidden = (';', '|', '&', '>', '<', '`', '$', '\\', '\n', '\r', '"', "'") + if any(token in raw_fallback for token in forbidden) or raw_fallback != raw_fallback.strip(): + issues.append("invalid_exact_fallback") + else: + parts = raw_fallback.split(" ") + if len(parts) != 4 or parts[0] != "context-guard-artifact" or parts[1] != "get" or parts[3] != "--full": + issues.append("invalid_exact_fallback") + elif SEMANTIC_GC_RECEIPT_ID_RE.fullmatch(parts[2]) is None: + issues.append("invalid_exact_fallback") + elif receipt_id is not None and parts[2] != receipt_id: + issues.append("fallback_receipt_mismatch") + elif receipt_id is None: + issues.append("invalid_exact_fallback") + else: + fallback = raw_fallback + + issues = ordered_semantic_gc_taxonomy(issues, SEMANTIC_GC_BLOCKER_ORDER) + return ({ + "candidate_replacement": None, + "candidate_safety_issues": issues, + "content_sha256": content_hash, + "exact_fallback_command": fallback, + "human_review_required": True, + "missed_context_note": note, + "protected_zone": False, + "provenance": {"receipt_id": receipt_id, "source_label": source_label}, + "reason": "unreachable_from_declared_roots", + "unit_id": unit["unit_id"], + }, issues, content_hash, receipt_id) + + +def semantic_gc_plan_payload(args: argparse.Namespace) -> dict[str, Any]: + raw_units = args.context_unit_json or [] + total_count = len(raw_units) + detailed_count = min(total_count, SEMANTIC_GC_DETAILED_UNIT_CAP) + overflow_count = max(total_count - SEMANTIC_GC_DETAILED_UNIT_CAP, 0) + units: list[dict[str, Any]] = [] + rows: list[dict[str, Any]] = [] + decoded_count = 0 + for index, raw in enumerate(raw_units[:SEMANTIC_GC_DETAILED_UNIT_CAP]): + decoded, row = decode_semantic_gc_unit(raw, index) + rows.append(row) + if decoded is None: + continue + decoded_count += 1 + units.append(normalize_semantic_gc_structure(decoded, row)) + + by_id: dict[str, list[dict[str, Any]]] = {} + for unit in units: + if unit["unit_id"] is not None: + by_id.setdefault(unit["unit_id"], []).append(unit) + duplicate_ids = {unit_id for unit_id, matches in by_id.items() if len(matches) > 1} + for unit in units: + structural = list(unit["row"]["structural_issues"]) + if unit["unit_id"] in duplicate_ids: + structural.append("duplicate_unit_id") + if any(reference not in by_id for reference in unit["reference_targets"]): + structural.append("unknown_reference") + if any(reference in duplicate_ids for reference in unit["reference_targets"]): + structural.append("ambiguous_reference") + unit["row"]["structural_issues"] = ordered_semantic_gc_taxonomy(structural, SEMANTIC_GC_BLOCKER_ORDER) + + declared_roots = sorted( + unit_id for unit_id, matches in by_id.items() + if unit_id not in duplicate_ids and len(matches) == 1 and matches[0]["is_root"] is True + ) + structural_blockers: list[str] = [] + if total_count == 0: + structural_blockers.append("no_context_units") + if overflow_count: + structural_blockers.append("unit_limit_exceeded") + for row in rows: + structural_blockers.extend(row["structural_issues"]) + if not declared_roots: + structural_blockers.append("no_declared_root") + graph_complete = not structural_blockers + + marked_ids: list[str] = [] + candidates: list[dict[str, Any]] = [] + unreachable_count = 0 + protected_unreachable_count = 0 + safety_valid_count = 0 + safety_invalid_count = 0 + safety_blockers: list[str] = [] + candidate_hashes: list[str] = [] + candidate_receipts: list[str] = [] + if graph_complete: + marked: set[str] = set() + pending = list(declared_roots) + while pending: + unit_id = pending.pop() + if unit_id in marked: + continue + marked.add(unit_id) + pending.extend(by_id[unit_id][0]["references"]) + marked_ids = sorted(marked) + for unit in units: + unit_id = unit["unit_id"] + assert unit_id is not None + row = unit["row"] + if unit_id in marked: + row["candidate_safety_applicable"] = False + continue + unreachable_count += 1 + if unit["protected_zone"] is True: + protected_unreachable_count += 1 + row["candidate_safety_applicable"] = False + continue + row["candidate_safety_applicable"] = True + candidate, issues, content_hash, receipt_id = normalize_semantic_gc_candidate(unit) + row["candidate_safety_issues"] = issues + candidates.append(candidate) + safety_blockers.extend(issues) + if issues: + safety_invalid_count += 1 + else: + safety_valid_count += 1 + if content_hash is not None: + candidate_hashes.append(content_hash) + if receipt_id is not None: + candidate_receipts.append(receipt_id) + candidates.sort(key=lambda item: item["unit_id"]) + + blockers = list(structural_blockers) + if not graph_complete: + blockers.append("graph_evaluation_suppressed") + protected_policy = "deny" if getattr(args, "protected_zone_policy", None) == "deny" else None + if protected_policy is None: + blockers.append("protected_zone_policy_required") + blockers.extend(safety_blockers) + if graph_complete and not args.provider_boundary_ack: + blockers.append("provider_boundary_ack_required") + if graph_complete and candidates and not args.human_review_ack: + blockers.append("human_review_ack_required") + blockers = ordered_semantic_gc_taxonomy(blockers, SEMANTIC_GC_BLOCKER_ORDER) + + warnings = list(SEMANTIC_GC_WARNING_ORDER[:6]) + if candidates: + warnings.extend(("human_review_still_required", "accepted_notes_are_untrusted")) + if len(candidate_hashes) != len(set(candidate_hashes)): + warnings.append("duplicate_content_sha256") + if len(candidate_receipts) != len(set(candidate_receipts)): + warnings.append("duplicate_receipt_id") + if protected_unreachable_count: + warnings.append("protected_unreachable_excluded") + if graph_complete and not candidates: + warnings.append("no_sweep_candidates") + warnings = ordered_semantic_gc_taxonomy(warnings, SEMANTIC_GC_WARNING_ORDER) + + verification_scope = { + "artifact_content_read": False, + "context_content_read": False, + "deletion_or_omission_performed": False, + "exact_fallback_executed": False, + "files_written": False, + "model_or_provider_called": False, + "network_called": False, + "provenance_verified_externally": False, + "subprocess_started": False, + } + return { + "blockers": blockers, + "candidate_count": len(candidates), + "candidate_replacement": None, + "candidate_safety_invalid_count": safety_invalid_count, + "candidate_safety_valid_count": safety_valid_count, + "candidates": candidates, + "declared_root_count": len(declared_roots), + "declared_root_ids": declared_roots, + "decoded_unit_count": decoded_count, + "detailed_unit_count": detailed_count, + "effective_protected_zone_policy": "deny", + "experiment": "semantic-gc", + "graph_evaluation_performed": graph_complete, + "graph_integrity_complete": graph_complete, + "human_review_acknowledged": bool(args.human_review_ack), + "human_review_performed": False, + "marked_unit_count": len(marked_ids), + "marked_unit_ids": marked_ids, + "omission_authorized": False, + "overflow_unit_count": overflow_count, + "plan_only": True, + "process_exit_contract": SEMANTIC_GC_PROCESS_EXIT_CONTRACT, + "protected_unreachable_count": protected_unreachable_count, + "protected_zone_policy": protected_policy, + "provider_boundary_acknowledged": bool(args.provider_boundary_ack), + "runtime_action_allowed": False, + "schema": SEMANTIC_GC_PLAN_SCHEMA_VERSION, + "status": "ready_for_plan_review" if not blockers else "blocked", + "structurally_valid_unit_count": sum(not row["structural_issues"] for row in rows), + "total_unit_count": total_count, + "unit_validation": rows, + "unreachable_unit_count": unreachable_count, + "verification_scope": verification_scope, + "warnings": warnings, + } + + +def command_plan_semantic_gc(args: argparse.Namespace) -> int: + payload = semantic_gc_plan_payload(args) + if args.json: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) + else: + print("ContextGuard semantic-gc plan") + print(f"Status: {payload['status']}") + print(f"Graph complete: {payload['graph_integrity_complete']}; candidates: {payload['candidate_count']}") + if payload["blockers"]: + print(f"Blockers: {', '.join(payload['blockers'])}") + print(f"Exit contract: {payload['process_exit_contract']}.") + print("semantic-gc is plan-only; no context was deleted, omitted, read, replaced, or authorized for runtime action.") + return 0 if payload["status"] == "ready_for_plan_review" else 2 + + def visual_crop_ocr_evidence_pack_payload(args: argparse.Namespace) -> dict[str, Any]: payload = visual_crop_ocr_plan_payload(args) blockers = list(payload["review_plan"]["readiness_blockers"]) @@ -5662,6 +6160,35 @@ def build_parser() -> argparse.ArgumentParser: proof_carrying_context.add_argument("--json", action="store_true", help="Emit JSON output.") proof_carrying_context.set_defaults(func=command_plan_proof_carrying_context) + semantic_gc = plan_sub.add_parser( + "semantic-gc", + allow_abbrev=False, + help="Plan caller-declared graph reachability candidates without reading or omitting context.", + ) + semantic_gc.add_argument( + "--context-unit-json", + action="append", + help="Inline literal semantic-GC unit JSON object. Repeatable; never treated as a path.", + ) + semantic_gc.add_argument( + "--provider-boundary-ack", + action="store_true", + help="Acknowledge that provider behavior and hosted savings remain unverified.", + ) + semantic_gc.add_argument( + "--human-review-ack", + action="store_true", + help="Acknowledge that candidate review remains required; this does not perform review.", + ) + semantic_gc.add_argument( + "--protected-zone-policy", + choices=("deny",), + default=None, + help="Explicit deny-only protected-zone declaration; omitted remains effective deny but blocks readiness.", + ) + semantic_gc.add_argument("--json", action="store_true", help="Emit JSON output.") + semantic_gc.set_defaults(func=command_plan_semantic_gc) + self_hosted = plan_sub.add_parser( "self-hosted-metrics-ledger", help="Dry-run self-hosted/local metrics ledger sidecar evidence without writing a ledger.", diff --git a/research/experimental-token-reduction-radar.md b/research/experimental-token-reduction-radar.md index bbccbb7..f416c5a 100644 --- a/research/experimental-token-reduction-radar.md +++ b/research/experimental-token-reduction-radar.md @@ -210,7 +210,13 @@ First local runtime experiment: `context-guard experiments record self-hosted-me - Does it preserve exact retrieval or clearly label lossy transforms? - Does it keep local/private data local unless the user explicitly opts into a provider call? +### Experimental semantic-GC plan gate + +`semantic-gc` is a default-off, deny-only, plan-review gate over a caller-declared graph. Default-off describes registry intent; the explicit plan CLI remains invocable and never enables omission or runtime action. Graph evaluation is suppressed when the complete envelope or topology is ambiguous. Unreachable nodes are review candidates, not proof of semantic irrelevance: omission and runtime action remain unauthorized. Candidate missed-context notes are untrusted. The planner does not read context/artifact content or verify provenance, fallback, providers, or hosted savings. Exit 0 means only `ready_for_plan_review`; it is never delete/omit authority. + +context-guard experiments plan semantic-gc --json --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false}' --context-unit-json '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"orphan","references":[],"is_root":false,"protected_zone":false,"content_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","provenance":{"source_label":"canonical-example","receipt_id":"0123456789abcdef"},"missed_context_note":"A reviewer could lose the orphaned rationale.","exact_fallback_command":"context-guard-artifact get 0123456789abcdef --full"}' --provider-boundary-ack --human-review-ack --protected-zone-policy deny + ## Current status -This radar is intentionally conservative. The shipped ContextGuard tools remain local context hygiene, artifact receipts, context packing, tool-schema pruning, transcript audit, statusline visibility, benchmark evidence, plus the explicitly gated local context-diff emitter, visual crop/OCR evidence-pack emitter, plan-only image-context-pack dry-run gate, plan-only/eval-only semantic-checkpoint gate, learned-compression candidate emitter, and self-hosted metrics sidecar recorder described above. Package-visible starter scaffolds live in [`../docs/experimental-benchmark-fixtures.md`](../docs/experimental-benchmark-fixtures.md). They are fixture-only synthetic task/variant examples and dry-run-only starters until prompts and success checks are replaced for a real experiment; they are not shipped runtime helpers, benchmark results, or hosted API savings evidence. +This radar is intentionally conservative. The shipped ContextGuard tools remain local context hygiene, artifact receipts, context packing, tool-schema pruning, transcript audit, statusline visibility, benchmark evidence, plus the explicitly gated local context-diff emitter, visual crop/OCR evidence-pack emitter, plan-only image-context-pack dry-run gate, plan-only/eval-only semantic-checkpoint gate, plan-only semantic-GC review gate, learned-compression candidate emitter, and self-hosted metrics sidecar recorder described above. Package-visible starter scaffolds live in [`../docs/experimental-benchmark-fixtures.md`](../docs/experimental-benchmark-fixtures.md). They are fixture-only synthetic task/variant examples and dry-run-only starters until prompts and success checks are replaced for a real experiment; they are not shipped runtime helpers, benchmark results, or hosted API savings evidence. Future experiments must pass the gates above before becoming product features. diff --git a/tests/test_context_guard_kit.py b/tests/test_context_guard_kit.py index 8b7bd09..96a1110 100644 --- a/tests/test_context_guard_kit.py +++ b/tests/test_context_guard_kit.py @@ -108,6 +108,8 @@ def load_command_manifest_for_tests(): PROOF_CARRYING_CONTEXT_RECEIPT = "0123456789abcdef" PROOF_CARRYING_CONTEXT_HASH = "a" * 64 PROOF_CARRYING_CONTEXT_TIMESTAMP = "2026-07-10T04:11:12Z" +SEMANTIC_GC_RECEIPT = "0123456789abcdef" +SEMANTIC_GC_HASH = "b" * 64 def proof_carrying_context_unit(**overrides) -> dict[str, object]: @@ -164,6 +166,50 @@ def run_proof_carrying_context_plan( ) +def semantic_gc_unit(unit_id: str, **overrides) -> dict[str, object]: + unit: dict[str, object] = { + "schema": "contextguard.semantic-gc-unit.v1", + "unit_id": unit_id, + "references": [], + "is_root": False, + "protected_zone": False, + } + unit.update(overrides) + return unit + + +def semantic_gc_candidate(unit_id: str = "orphan", **overrides) -> dict[str, object]: + unit = semantic_gc_unit( + unit_id, + content_sha256=SEMANTIC_GC_HASH, + provenance={"source_label": "canonical-example", "receipt_id": SEMANTIC_GC_RECEIPT}, + missed_context_note="A reviewer could lose the orphaned rationale.", + exact_fallback_command=f"context-guard-artifact get {SEMANTIC_GC_RECEIPT} --full", + ) + unit.update(overrides) + return unit + + +def run_semantic_gc_plan( + script: Path, + units: list[object] | None = None, + *, + provider_boundary_ack: bool = True, + human_review_ack: bool = True, + protected_zone_policy: str | None = "deny", +) -> subprocess.CompletedProcess[str]: + argv = [sys.executable, str(script), "plan", "semantic-gc", "--json"] + for unit in units if units is not None else [semantic_gc_unit("root", is_root=True), semantic_gc_candidate()]: + argv.extend(["--context-unit-json", unit if isinstance(unit, str) else json.dumps(unit)]) + if provider_boundary_ack: + argv.append("--provider-boundary-ack") + if human_review_ack: + argv.append("--human-review-ack") + if protected_zone_policy is not None: + argv.extend(["--protected-zone-policy", protected_zone_policy]) + return subprocess.run(argv, text=True, capture_output=True) + + def reserve_loopback_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("127.0.0.1", 0)) @@ -4075,6 +4121,493 @@ def test_experimental_proof_carrying_context_parser_and_human_surface_are_narrow self.assertIn("No source/artifact/config/stdin content was read", proc.stdout) self.assertIn("no context was generated or replaced", proc.stdout) + def test_experimental_semantic_gc_registry_and_surface(self): + for script in EXPERIMENT_SCRIPTS: + with self.subTest(script=script): + registry = json.loads(subprocess.run( + [sys.executable, str(script), "list", "--json"], + text=True, capture_output=True, check=True, + ).stdout) + row = next(item for item in registry["experiments"] if item["id"] == "semantic-gc") + self.assertFalse(row["default_enabled"]) + self.assertEqual(row["risk_level"], "high") + self.assertEqual(row["runtime_status"], "available-plan-only") + self.assertEqual(row["commands"], ["context-guard experiments plan semantic-gc"]) + self.assertIn( + "provider-boundary acknowledgement for every complete graph; " + "human-review acknowledgement when unprotected sweep candidates exist", + row["gate_requirements"], + ) + self.assertIn( + "Ready plans exit 0; blocked plans still emit their envelope and exit 2.", + row["evidence_contract"], + ) + module = load_python_script_module(script, f"_semantic_gc_parser_{script.name.replace('-', '_')}") + parser = module.build_parser() + root_sub = next(action for action in parser._actions if isinstance(action, argparse._SubParsersAction)) + plan = root_sub.choices["plan"] + plan_sub = next(action for action in plan._actions if isinstance(action, argparse._SubParsersAction)) + semantic = plan_sub.choices["semantic-gc"] + self.assertEqual( + {option for action in semantic._actions for option in action.option_strings if option not in {"-h", "--help"}}, + {"--context-unit-json", "--provider-boundary-ack", "--human-review-ack", "--protected-zone-policy", "--json"}, + ) + payload = json.loads(run_semantic_gc_plan(script).stdout) + self.assertEqual(payload["experiment"], "semantic-gc") + for verb in ("emit", "record", "serve", "apply", "delete", "omit"): + proc = subprocess.run([sys.executable, str(script), verb, "semantic-gc"], text=True, capture_output=True) + self.assertEqual(proc.returncode, 2) + + def test_experimental_semantic_gc_ready_for_plan_review_contract(self): + expected_warnings = [ + "plan_only_no_omission", "caller_declared_graph_unverified", + "semantic_relevance_not_evaluated", "provider_boundary_not_verified", + "provenance_not_verified_externally", "fallback_not_executed", + "human_review_still_required", "accepted_notes_are_untrusted", + ] + for script in EXPERIMENT_SCRIPTS: + first = run_semantic_gc_plan(script) + second = run_semantic_gc_plan(script) + self.assertEqual(first.returncode, 0) + self.assertEqual(second.returncode, 0) + self.assertEqual(first.stdout, second.stdout) + payload = json.loads(first.stdout) + self.assertEqual(payload["schema"], "contextguard.experiments.semantic-gc-plan.v1") + self.assertEqual(payload["status"], "ready_for_plan_review") + self.assertTrue(payload["graph_integrity_complete"]) + self.assertTrue(payload["graph_evaluation_performed"]) + self.assertEqual(payload["declared_root_ids"], ["root"]) + self.assertEqual(payload["marked_unit_ids"], ["root"]) + self.assertEqual(payload["candidate_count"], 1) + self.assertEqual(payload["candidates"][0]["unit_id"], "orphan") + self.assertEqual(payload["warnings"], expected_warnings) + self.assertEqual(payload["blockers"], []) + self.assertIsNone(payload["candidate_replacement"]) + self.assertFalse(payload["human_review_performed"]) + self.assertFalse(payload["omission_authorized"]) + self.assertFalse(payload["runtime_action_allowed"]) + self.assertEqual( + payload["process_exit_contract"], + "exit code 0 means ready_for_plan_review; exit code 2 means a blocked plan was emitted", + ) + self.assertEqual(first.stdout, json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n") + + def test_experimental_semantic_gc_graph_integrity_suppression(self): + fixtures = [ + ([], "no_context_units"), + ([semantic_gc_unit("root", is_root=True, references=["missing"])], "unknown_reference"), + ([semantic_gc_unit("dup", is_root=True), semantic_gc_unit("dup")], "duplicate_unit_id"), + ([semantic_gc_unit("root", is_root="yes")], "invalid_root_flag"), + ([semantic_gc_unit("root", is_root=True, protected_zone="yes")], "invalid_protected_zone_flag"), + ([semantic_gc_unit("lonely")], "no_declared_root"), + ] + for script in EXPERIMENT_SCRIPTS: + for units, issue in fixtures: + proc = run_semantic_gc_plan(script, units) + self.assertEqual(proc.returncode, 2) + payload = json.loads(proc.stdout) + self.assertIn(issue, payload["blockers"]) + self.assertIn("graph_evaluation_suppressed", payload["blockers"]) + self.assertFalse(payload["graph_integrity_complete"]) + self.assertFalse(payload["graph_evaluation_performed"]) + self.assertEqual(payload["marked_unit_ids"], []) + self.assertEqual(payload["candidates"], []) + self.assertEqual(payload["candidate_count"], 0) + self.assertTrue(all(row["candidate_safety_applicable"] is None for row in payload["unit_validation"])) + + def test_experimental_semantic_gc_two_phase_validation(self): + bad = semantic_gc_candidate(content_sha256="BAD", provenance=None, missed_context_note=None, exact_fallback_command=None) + graph = [semantic_gc_unit("root", is_root=True, references=["live"]), semantic_gc_unit("live"), bad] + for script in EXPERIMENT_SCRIPTS: + proc = run_semantic_gc_plan(script, graph) + self.assertEqual(proc.returncode, 2) + payload = json.loads(proc.stdout) + self.assertTrue(payload["graph_evaluation_performed"]) + self.assertEqual(payload["marked_unit_ids"], ["live", "root"]) + self.assertEqual(payload["candidate_count"], 1) + self.assertEqual(payload["candidates"][0]["unit_id"], "orphan") + self.assertEqual(payload["candidate_safety_invalid_count"], 1) + self.assertIsNone(payload["candidates"][0]["content_sha256"]) + self.assertIn("invalid_content_sha256", payload["blockers"]) + self.assertIn("missing_provenance", payload["blockers"]) + self.assertEqual(payload["status"], "blocked") + self.assertFalse(payload["human_review_performed"]) + self.assertFalse(payload["omission_authorized"]) + self.assertFalse(payload["runtime_action_allowed"]) + + def test_experimental_semantic_gc_counts_and_audit(self): + graph = [ + semantic_gc_unit("root", is_root=True, references=["a"]), + semantic_gc_unit("a", references=["root"]), + semantic_gc_candidate("z"), + semantic_gc_candidate("p", protected_zone=True), + ] + for script in EXPERIMENT_SCRIPTS: + payload = json.loads(run_semantic_gc_plan(script, graph).stdout) + self.assertEqual(payload["total_unit_count"], 4) + self.assertEqual(payload["decoded_unit_count"], 4) + self.assertEqual(payload["structurally_valid_unit_count"], 4) + self.assertEqual(payload["marked_unit_ids"], ["a", "root"]) + self.assertEqual(payload["unreachable_unit_count"], 2) + self.assertEqual(payload["protected_unreachable_count"], 1) + self.assertEqual(payload["candidate_count"], 1) + self.assertEqual([row["input_index"] for row in payload["unit_validation"]], [0, 1, 2, 3]) + self.assertIn("protected_unreachable_excluded", payload["warnings"]) + + def test_experimental_semantic_gc_unreachable_cycle_becomes_candidates(self): + graph = [ + semantic_gc_unit("root", is_root=True), + semantic_gc_candidate("cycle-a", references=["cycle-b"]), + semantic_gc_candidate("cycle-b", references=["cycle-a"], content_sha256="c" * 64, + provenance={"source_label": "cycle-b", "receipt_id": "abcdef0123456789"}, + exact_fallback_command="context-guard-artifact get abcdef0123456789 --full"), + ] + for script in EXPERIMENT_SCRIPTS: + payload = json.loads(run_semantic_gc_plan(script, graph).stdout) + self.assertTrue(payload["graph_evaluation_performed"]) + self.assertEqual(payload["marked_unit_ids"], ["root"]) + self.assertEqual(payload["unreachable_unit_count"], 2) + self.assertEqual(payload["candidate_count"], 2) + self.assertEqual([row["unit_id"] for row in payload["candidates"]], ["cycle-a", "cycle-b"]) + self.assertEqual(payload["candidate_safety_valid_count"], 2) + self.assertEqual(payload["status"], "ready_for_plan_review") + + def test_experimental_semantic_gc_input_permutation_preserves_canonical_graph_contract(self): + graph = [ + semantic_gc_unit("root", is_root=True, references=["live"]), + semantic_gc_unit("live"), + semantic_gc_candidate("z-candidate", content_sha256="c" * 64, + provenance={"source_label": "z", "receipt_id": "abcdef0123456789"}, + exact_fallback_command="context-guard-artifact get abcdef0123456789 --full"), + semantic_gc_candidate("a-candidate"), + ] + + def normalized(payload): + result = dict(payload) + result["unit_validation"] = sorted( + ({key: value for key, value in row.items() if key != "input_index"} + for row in payload["unit_validation"]), + key=lambda row: row["unit_id"], + ) + return result + + for script in EXPERIMENT_SCRIPTS: + forward = json.loads(run_semantic_gc_plan(script, graph).stdout) + reverse = json.loads(run_semantic_gc_plan(script, list(reversed(graph))).stdout) + self.assertEqual(normalized(forward), normalized(reverse)) + self.assertNotEqual( + [row["unit_id"] for row in forward["unit_validation"]], + [row["unit_id"] for row in reverse["unit_validation"]], + ) + + def test_experimental_semantic_gc_protected_only_omitted_policy_still_evaluates(self): + graph = [ + semantic_gc_unit("root", is_root=True), + semantic_gc_unit("protected", protected_zone=True), + ] + for script in EXPERIMENT_SCRIPTS: + payload = json.loads(run_semantic_gc_plan(script, graph, protected_zone_policy=None).stdout) + self.assertTrue(payload["graph_integrity_complete"]) + self.assertTrue(payload["graph_evaluation_performed"]) + self.assertEqual(payload["candidate_count"], 0) + self.assertEqual(payload["protected_unreachable_count"], 1) + self.assertIsNone(payload["protected_zone_policy"]) + self.assertEqual(payload["effective_protected_zone_policy"], "deny") + self.assertEqual(payload["blockers"], ["protected_zone_policy_required"]) + self.assertEqual(payload["status"], "blocked") + + def test_experimental_semantic_gc_depth_100_allowed_and_101_rejected(self): + def raw_with_max_depth(max_depth): + nested: object = 0 + for _ in range(max_depth - 1): + nested = {"x": nested} + return json.dumps(semantic_gc_unit("root", is_root=True, provenance=nested)) + + for script in EXPERIMENT_SCRIPTS: + allowed = json.loads(run_semantic_gc_plan(script, [raw_with_max_depth(100)]).stdout) + self.assertNotIn("context_unit_depth_exceeded", allowed["blockers"]) + self.assertTrue(allowed["graph_evaluation_performed"]) + rejected = json.loads(run_semantic_gc_plan(script, [raw_with_max_depth(101)]).stdout) + self.assertIn("context_unit_depth_exceeded", rejected["blockers"]) + self.assertFalse(rejected["graph_evaluation_performed"]) + + def test_experimental_semantic_gc_candidate_safety(self): + invalid_note = " SECRET\n" + cases = [ + ({"content_sha256": "A" * 64}, "invalid_content_sha256"), + ({"provenance": {"source_label": " bad ", "receipt_id": SEMANTIC_GC_RECEIPT}}, "invalid_source_label"), + ({"provenance": {"source_label": "ok", "receipt_id": "BADSECRET"}}, "invalid_receipt_id"), + ({"provenance": {"source_label": "ok", "receipt_id": "a" * 65}, + "exact_fallback_command": f"context-guard-artifact get {'a' * 65} --full"}, "invalid_receipt_id"), + ({"missed_context_note": invalid_note}, "invalid_missed_context_note"), + ({"missed_context_note": "SAFE\u2028SECRET"}, "invalid_missed_context_note"), + ({"missed_context_note": "SAFE\u202eSECRET"}, "invalid_missed_context_note"), + ({"exact_fallback_command": "context-guard-artifact get deadbeefdeadbeef --full"}, "fallback_receipt_mismatch"), + ({"exact_fallback_command": f"context-guard-artifact get {SEMANTIC_GC_RECEIPT} --full; echo SECRET"}, "invalid_exact_fallback"), + ] + for script in EXPERIMENT_SCRIPTS: + for overrides, blocker in cases: + proc = run_semantic_gc_plan(script, [semantic_gc_unit("root", is_root=True), semantic_gc_candidate(**overrides)]) + self.assertEqual(proc.returncode, 2) + payload = json.loads(proc.stdout) + self.assertIn(blocker, payload["blockers"]) + self.assertEqual(payload["candidate_count"], 1) + self.assertEqual(payload["candidate_safety_invalid_count"], 1) + self.assertEqual(payload["status"], "blocked") + self.assertFalse(payload["human_review_performed"]) + self.assertFalse(payload["omission_authorized"]) + self.assertFalse(payload["runtime_action_allowed"]) + self.assertNotIn("BADSECRET", payload["candidates"][0].__repr__()) + self.assertNotIn("SECRET", json.dumps(payload)) + + def test_experimental_semantic_gc_acknowledgement_invariants(self): + for script in EXPERIMENT_SCRIPTS: + for provider, human, blocker in ((False, True, "provider_boundary_ack_required"), (True, False, "human_review_ack_required")): + proc = run_semantic_gc_plan(script, provider_boundary_ack=provider, human_review_ack=human) + self.assertEqual(proc.returncode, 2) + payload = json.loads(proc.stdout) + self.assertIn(blocker, payload["blockers"]) + self.assertEqual(payload["status"], "blocked") + self.assertEqual( + payload["process_exit_contract"], + "exit code 0 means ready_for_plan_review; exit code 2 means a blocked plan was emitted", + ) + self.assertFalse(payload["human_review_performed"]) + self.assertFalse(payload["omission_authorized"]) + self.assertFalse(payload["runtime_action_allowed"]) + + no_candidate = json.loads(run_semantic_gc_plan( + script, + [semantic_gc_unit("root", is_root=True)], + provider_boundary_ack=True, + human_review_ack=False, + ).stdout) + self.assertEqual(no_candidate["candidate_count"], 0) + self.assertNotIn("human_review_ack_required", no_candidate["blockers"]) + self.assertEqual(no_candidate["status"], "ready_for_plan_review") + self.assertFalse(no_candidate["human_review_acknowledged"]) + self.assertFalse(no_candidate["human_review_performed"]) + self.assertFalse(no_candidate["omission_authorized"]) + self.assertFalse(no_candidate["runtime_action_allowed"]) + + def test_experimental_semantic_gc_omitted_protected_policy(self): + graph = [semantic_gc_unit("root", is_root=True), semantic_gc_candidate("orphan"), semantic_gc_unit("protected", protected_zone=True)] + for script in EXPERIMENT_SCRIPTS: + proc = run_semantic_gc_plan(script, graph, protected_zone_policy=None) + self.assertEqual(proc.returncode, 2) + payload = json.loads(proc.stdout) + self.assertIsNone(payload["protected_zone_policy"]) + self.assertEqual(payload["effective_protected_zone_policy"], "deny") + self.assertTrue(payload["graph_evaluation_performed"]) + self.assertEqual(payload["protected_unreachable_count"], 1) + self.assertEqual(payload["candidate_count"], 1) + self.assertIn("protected_zone_policy_required", payload["blockers"]) + self.assertEqual(payload["status"], "blocked") + self.assertFalse(payload["omission_authorized"]) + self.assertFalse(payload["runtime_action_allowed"]) + + unprotected_only = run_semantic_gc_plan( + script, + [semantic_gc_unit("root", is_root=True)], + protected_zone_policy=None, + ) + self.assertEqual(unprotected_only.returncode, 2) + unprotected_payload = json.loads(unprotected_only.stdout) + self.assertEqual(unprotected_payload["blockers"], ["protected_zone_policy_required"]) + self.assertEqual(unprotected_payload["status"], "blocked") + + def test_experimental_semantic_gc_direct_invalid_protected_policy_fails_closed(self): + raw_root = json.dumps(semantic_gc_unit("root", is_root=True)) + for index, script in enumerate(EXPERIMENT_SCRIPTS): + module = load_python_script_module(script, f"_semantic_gc_policy_{index}") + for invalid_policy in ("allow", "DENY", " deny ", ""): + with self.subTest(script=script, policy=invalid_policy): + payload = module.semantic_gc_plan_payload(argparse.Namespace( + context_unit_json=[raw_root], + provider_boundary_ack=True, + human_review_ack=True, + protected_zone_policy=invalid_policy, + )) + self.assertIsNone(payload["protected_zone_policy"]) + self.assertEqual(payload["effective_protected_zone_policy"], "deny") + self.assertIn("protected_zone_policy_required", payload["blockers"]) + self.assertEqual(payload["status"], "blocked") + + cli = subprocess.run( + [ + sys.executable, + str(script), + "plan", + "semantic-gc", + "--json", + "--context-unit-json", + raw_root, + "--provider-boundary-ack", + "--human-review-ack", + "--protected-zone-policy", + invalid_policy, + ], + text=True, + capture_output=True, + ) + self.assertEqual(cli.returncode, 2) + self.assertEqual(cli.stdout, "") + self.assertIn("invalid choice", cli.stderr) + + def test_experimental_semantic_gc_mixed_reference_defects_are_all_reported(self): + graph = [ + semantic_gc_unit( + "root", + is_root=True, + references=["missing", "missing", 42], + ) + ] + expected = ["invalid_references", "duplicate_reference", "unknown_reference"] + for script in EXPERIMENT_SCRIPTS: + payload = json.loads(run_semantic_gc_plan(script, graph).stdout) + self.assertEqual(payload["unit_validation"][0]["structural_issues"], expected) + for issue in expected: + self.assertIn(issue, payload["blockers"]) + self.assertFalse(payload["graph_evaluation_performed"]) + + def test_experimental_semantic_gc_duplicate_target_is_ambiguous_not_unknown(self): + graph = [ + semantic_gc_unit("root", is_root=True, references=["dup"]), + semantic_gc_unit("dup"), + semantic_gc_unit("dup"), + ] + for script in EXPERIMENT_SCRIPTS: + proc = run_semantic_gc_plan(script, graph) + self.assertEqual(proc.returncode, 2) + payload = json.loads(proc.stdout) + self.assertIn("duplicate_unit_id", payload["blockers"]) + self.assertIn("ambiguous_reference", payload["blockers"]) + self.assertNotIn("unknown_reference", payload["unit_validation"][0]["structural_issues"]) + self.assertFalse(payload["graph_evaluation_performed"]) + + def test_experimental_semantic_gc_rejects_argparse_prefix_abbreviations(self): + root = json.dumps(semantic_gc_unit("root", is_root=True)) + abbreviated_flags = ( + ("--context-unit-j", root), + ("--provider-boundary-a", None), + ("--human-review-a", None), + ("--protected-zone-p", "deny"), + ) + for script in EXPERIMENT_SCRIPTS: + for flag, value in abbreviated_flags: + with self.subTest(script=script, flag=flag): + argv = [sys.executable, str(script), "plan", "semantic-gc", flag] + if value is not None: + argv.append(value) + proc = subprocess.run(argv, text=True, capture_output=True) + self.assertEqual(proc.returncode, 2) + self.assertIn("unrecognized arguments", proc.stderr) + + def test_experimental_semantic_gc_strict_envelope_ambiguity(self): + marked = '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false,"provenance":{"source_label":"a","source_label":"b","receipt_id":"0123456789abcdef"}}' + protected = '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"protected","references":[],"is_root":false,"protected_zone":true,"provenance":{"source_label":"a","source_label":"b","receipt_id":"0123456789abcdef"}}' + for script in EXPERIMENT_SCRIPTS: + for units in ([marked], [json.dumps(semantic_gc_unit("root", is_root=True)), protected]): + payload = json.loads(run_semantic_gc_plan(script, units).stdout) + self.assertIn("duplicate_json_key", payload["blockers"]) + self.assertFalse(payload["graph_evaluation_performed"]) + self.assertEqual(payload["candidates"], []) + + def test_experimental_semantic_gc_strict_json_and_non_echo(self): + def nested(depth): + value: object = 0 + for _ in range(depth): + value = {"x": value} + return json.dumps(value) + fixtures = [ + ("{", "invalid_context_unit_json"), + ('{"schema":"contextguard.semantic-gc-unit.v1","schema":"SECRET"}', "duplicate_json_key"), + (nested(101), "context_unit_depth_exceeded"), + ('{"x":NaN}', "nonfinite_json_number"), + (json.dumps({"x": "\ud800"}), "invalid_unicode_scalar"), + ('{"x":"' + ("a" * 8190) + '"}', "invalid_context_unit_json"), + ] + for script in EXPERIMENT_SCRIPTS: + for raw, blocker in fixtures: + payload = json.loads(run_semantic_gc_plan(script, [raw]).stdout) + self.assertIn(blocker, payload["blockers"]) + self.assertNotIn("SECRET", json.dumps(payload)) + exact = '{"schema":"contextguard.semantic-gc-unit.v1","unit_id":"root","references":[],"is_root":true,"protected_zone":false,"extra":"' + ("a" * 8060) + '"}' + exact = exact[:-2] + ("a" * (8192 - len(exact.encode("utf-8")))) + '"}' + self.assertEqual(len(exact.encode("utf-8")), 8192) + payload = json.loads(run_semantic_gc_plan(script, [exact]).stdout) + self.assertIn("unknown_context_unit_field", payload["blockers"]) + payload = json.loads(run_semantic_gc_plan(script, [exact + " "]).stdout) + self.assertIn("invalid_context_unit_json", payload["blockers"]) + safe_raw = json.dumps(semantic_gc_unit("root", is_root=True)) + deep_100 = nested(100) + module = load_python_script_module(script, f"_semantic_gc_recursion_{script.name.replace('-', '_')}") + with mock.patch.object(module.json, "loads", side_effect=RecursionError): + payload = module.semantic_gc_plan_payload(argparse.Namespace( + context_unit_json=[safe_raw], provider_boundary_ack=True, + human_review_ack=True, protected_zone_policy="deny", + )) + self.assertIn("decoder_recursion_limit", payload["blockers"]) + duplicate_precedence = '{"x":' + deep_100 + ',"x":NaN}' + payload = json.loads(run_semantic_gc_plan(script, [duplicate_precedence]).stdout) + self.assertIn("duplicate_json_key", payload["blockers"]) + self.assertNotIn("nonfinite_json_number", payload["blockers"]) + + secret = "DO_NOT_ECHO_SEMANTIC_GC_SECRET" + proc = run_semantic_gc_plan(script, [json.dumps(semantic_gc_unit("root", is_root=True, extra=secret))]) + self.assertNotIn(secret, proc.stdout + proc.stderr) + + human = subprocess.run( + [sys.executable, str(script), "plan", "semantic-gc", + "--context-unit-json", json.dumps(semantic_gc_unit("root", is_root=True)), + "--context-unit-json", json.dumps(semantic_gc_candidate(missed_context_note=secret)), + "--provider-boundary-ack", "--human-review-ack", "--protected-zone-policy", "deny"], + text=True, capture_output=True, check=True, + ) + self.assertNotIn(secret, human.stdout + human.stderr) + + def test_experimental_semantic_gc_overflow_and_side_effects(self): + class Exploding: + def boom(self, *_args, **_kwargs): raise AssertionError("overflow touched") + __str__ = __repr__ = __eq__ = __hash__ = boom + encode = boom + raw = json.dumps(semantic_gc_unit("root", is_root=True)) + for index, script in enumerate(EXPERIMENT_SCRIPTS): + module = load_python_script_module(script, f"_semantic_gc_{index}") + payload = module.semantic_gc_plan_payload(argparse.Namespace( + context_unit_json=[raw] * 64 + [Exploding()], provider_boundary_ack=True, + human_review_ack=True, protected_zone_policy="deny", + )) + self.assertEqual(payload["total_unit_count"], 65) + self.assertEqual(payload["overflow_unit_count"], 1) + self.assertIn("unit_limit_exceeded", payload["blockers"]) + scope = payload["verification_scope"] + self.assertTrue(all(value is False for value in scope.values())) + candidate_raw = json.dumps(semantic_gc_candidate()) + + def forbidden(name): + def fail(*_args, **_kwargs): + raise AssertionError(f"forbidden semantic-gc side effect: {name}") + return fail + + with contextlib.ExitStack() as stack: + stack.enter_context(mock.patch("builtins.open", side_effect=forbidden("open"))) + for name in ("open", "read_text", "read_bytes", "write_text", "write_bytes"): + stack.enter_context(mock.patch.object(Path, name, side_effect=forbidden(f"Path.{name}"))) + for name in ("system", "popen", "mkdir", "makedirs", "remove", "unlink", "rename", "replace"): + stack.enter_context(mock.patch.object(module.os, name, side_effect=forbidden(f"os.{name}"))) + for name in ("time", "monotonic", "perf_counter", "sleep"): + stack.enter_context(mock.patch.object(module.time, name, side_effect=forbidden(f"time.{name}"))) + for name in ("socket", "create_connection", "getaddrinfo", "gethostbyname"): + stack.enter_context(mock.patch.object(module.socket, name, side_effect=forbidden(f"socket.{name}"))) + for name in ("run", "Popen", "call", "check_call", "check_output"): + stack.enter_context(mock.patch.object(subprocess, name, side_effect=forbidden(f"subprocess.{name}"))) + ready = module.semantic_gc_plan_payload(argparse.Namespace( + context_unit_json=[raw, candidate_raw], provider_boundary_ack=True, + human_review_ack=True, protected_zone_policy="deny", + )) + self.assertEqual(ready["status"], "ready_for_plan_review") + def test_experimental_visual_crop_ocr_emit_runtime(self): for script in EXPERIMENT_SCRIPTS: with self.subTest(script=script): @@ -8153,6 +8686,31 @@ def test_experimental_proof_carrying_context_docs_examples_match_cli(self): self.assertEqual(payload["review_plan"]["warnings"], expected_warnings) self.assertIsNone(payload["candidate_replacement"]) + def test_experimental_semantic_gc_docs_parity(self): + docs = ( + ROOT / "README.md", ROOT / "README.ko.md", KIT_DIR / "README.md", + PLUGIN_DIR / "README.md", PLUGIN_DIR / "README.ko.md", + ROOT / "research" / "experimental-token-reduction-radar.md", + ) + for doc in docs: + text = doc.read_text(encoding="utf-8") + lower = text.lower() + self.assertEqual(sum(line.strip().startswith("context-guard experiments plan semantic-gc") for line in text.splitlines()), 1) + self.assertIn("graph", lower) + self.assertRegex(lower, r"suppress|억제") + self.assertRegex(lower, r"plan.review|계획 검토") + self.assertRegex(lower, r"untrusted|신뢰되지") + self.assertRegex(lower, r"deny.only|deny 전용") + self.assertRegex(lower, r"does not (?:read|verify)|읽지 않|검증하지 않") + self.assertIn("registry intent", lower) + self.assertIn("exit 0", lower) + self.assertRegex(lower, r"never delete/omit authority|delete/omit 권한이 아") + line = next(line.strip() for line in text.splitlines() if line.strip().startswith("context-guard experiments plan semantic-gc")) + argv = shlex.split(line) + proc = subprocess.run([sys.executable, str(PLUGIN_BIN / "context-guard-experiments"), *argv[2:]], text=True, capture_output=True, check=True) + payload = json.loads(proc.stdout) + self.assertEqual(payload["status"], "ready_for_plan_review") + def test_experimental_local_proxy_docs_surface_boundary(self): docs = ( ROOT / "README.md",