Research-Stack/tools/kotc/kotc_sim.py

#!/usr/bin/env python3
"""
KOTC — Kinetic Operation Token Completion Daemon simulator.

This is a local prototype scaffold. It does not call a real model. It treats a
candidate completion as an auditable operation and emits a receipt.

Usage:
  python tools/kotc/kotc_sim.py --candidate candidate.txt --target Semantics.InvariantReceipt.Core --mode REPAIR
  python tools/kotc/kotc_sim.py --candidate - --target docs/research/KOTC --mode DRAFT < candidate.txt
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
from dataclasses import dataclass, asdict
from enum import Enum
from pathlib import Path
from typing import Dict, List, Tuple


class Mode(str, Enum):
    DRAFT = "DRAFT"
    REPAIR = "REPAIR"
    STRICT = "STRICT"


class Decision(str, Enum):
    ACCEPT = "ACCEPT"
    REJECT = "REJECT"
    HOLD = "HOLD"
    QUARANTINE = "QUARANTINE"


class Risk(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass(frozen=True)
class CompletionReceipt:
    receipt_type: str
    completion_id: str
    mode: str
    target_module: str
    context_files: List[str]
    symbol_refs: List[str]
    kot_cost_q16: str
    risk: str
    decision: str
    policy_checks: Dict[str, str]
    candidate_hash: str
    notes: List[str]


def sha256_text(text: str) -> str:
    return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()


def q16_hex(units: int) -> str:
    """Encode integer units as Q16.16 raw hex."""
    raw = max(0, min(units << 16, 0xFFFFFFFF))
    return f"0x{raw:08X}"


def estimate_kot_units(candidate: str, mode: Mode, policy_failures: int) -> int:
    base = max(1, len(candidate.encode("utf-8")) // 128)
    mode_multiplier = {
        Mode.DRAFT: 1,
        Mode.REPAIR: 2,
        Mode.STRICT: 4,
    }[mode]
    return base * mode_multiplier + policy_failures * 16


def extract_symbol_refs(candidate: str) -> List[str]:
    refs = set(re.findall(r"\b[A-Z][A-Za-z0-9_]{2,}\b", candidate))
    allow = {
        "Q16_16",
        "Receipt",
        "Outcome",
        "ModelUpgrade",
        "SubstrateAdapter",
        "RegisteredModel",
        "KOT",
        "AMMR",
        "AVMR",
        "NUVMAP",
        "GCCL",
        "GCCLRep",
        "InvariantReceipt",
    }
    return sorted(ref for ref in refs if ref in allow)


def check_policy(candidate: str, mode: Mode, target: str) -> Tuple[Dict[str, str], List[str]]:
    notes: List[str] = []
    checks: Dict[str, str] = {}

    hot_path_target = any(token.lower() in target.lower() for token in [
        "fixedpoint", "physics", "kernel", "codec", "hot", "semantics"
    ])
    float_patterns = [r"\bFloat\b", r"\bDouble\b", r"\bf32\b", r"\bf64\b", r"\bfloat\b"]
    has_float = any(re.search(pattern, candidate) for pattern in float_patterns)
    if hot_path_target and has_float:
        checks["no_float_hot_path"] = "failed"
        notes.append("Float-like type or token appears in a hot-path/core target.")
    elif has_float:
        checks["no_float_hot_path"] = "warning"
        notes.append("Float-like token appears outside an obvious hot path; review required.")
    else:
        checks["no_float_hot_path"] = "passed"

    physical_claim_tokens = ["SI", "joule", "newton", "mass", "energy", "thermodynamic", "physics"]
    has_physical_claim = any(token.lower() in candidate.lower() for token in physical_claim_tokens)
    has_receipt_or_scope = any(token.lower() in candidate.lower() for token in ["receipt", "dimensionless", "proxy", "not_si", "not physical"])
    if has_physical_claim and not has_receipt_or_scope:
        checks["no_unreviewed_physical_claim"] = "failed"
        notes.append("Physical/SI-like claim appears without receipt/scope/dimensionless boundary.")
    elif has_physical_claim:
        checks["no_unreviewed_physical_claim"] = "warning"
    else:
        checks["no_unreviewed_physical_claim"] = "passed"

    theorem_weakening_patterns = [r"admit", r"axiom\s+", r"unsafe", r"set_option\s+autoImplicit\s+true"]
    has_theorem_weakening = any(re.search(pattern, candidate, re.IGNORECASE) for pattern in theorem_weakening_patterns)
    if has_theorem_weakening:
        checks["no_theorem_weakening"] = "failed"
        notes.append("Candidate contains proof-weakening or unsafe pattern.")
    else:
        checks["no_theorem_weakening"] = "passed"

    introduces_invariant = bool(re.search(r"\bInvariant\b|\binvariant\b|\btheorem\b|\blemma\b", candidate))
    has_receipt = "Receipt" in candidate or "receipt" in candidate.lower()
    if introduces_invariant and not has_receipt and mode == Mode.STRICT:
        checks["no_unchecked_invariant_introduction"] = "failed"
        notes.append("Strict-mode invariant/theorem introduction lacks receipt language.")
    elif introduces_invariant and not has_receipt:
        checks["no_unchecked_invariant_introduction"] = "warning"
    else:
        checks["no_unchecked_invariant_introduction"] = "passed"

    compiler_pass_tokens = ["CompilerPass", "pass_id", "workflow", "WorkflowDAG", "transform"]
    mentions_pass = any(token in candidate for token in compiler_pass_tokens)
    mentions_cost = any(token.lower() in candidate.lower() for token in ["cost", "kot", "budget"])
    if mentions_pass and not mentions_cost:
        checks["no_unbudgeted_compiler_pass"] = "failed"
        notes.append("Compiler-pass-like candidate lacks KOT/cost/budget accounting.")
    else:
        checks["no_unbudgeted_compiler_pass"] = "passed"

    authority_tokens = ["coreModule", "CORE_MODULE", "proved", "certified", "ASIL-D", "canonical"]
    has_authority = any(token in candidate for token in authority_tokens)
    if has_authority and not has_receipt:
        checks["no_authority_escalation"] = "failed"
        notes.append("Candidate escalates authority/certification without receipt evidence.")
    elif has_authority:
        checks["no_authority_escalation"] = "warning"
    else:
        checks["no_authority_escalation"] = "passed"

    if "Receipt" not in candidate and "receipt" not in candidate.lower():
        checks["no_silent_receipt_drop"] = "warning" if mode != Mode.DRAFT else "not_applicable"
    else:
        checks["no_silent_receipt_drop"] = "passed"

    if re.search(r"while\s+true|partial\s+def|rec\s+", candidate):
        checks["no_unbounded_recursion"] = "warning"
        notes.append("Potentially unbounded recursion/loop pattern needs review.")
    else:
        checks["no_unbounded_recursion"] = "passed"

    return checks, notes


def decide(checks: Dict[str, str], mode: Mode) -> Tuple[Decision, Risk]:
    failures = sum(1 for value in checks.values() if value == "failed")
    warnings = sum(1 for value in checks.values() if value == "warning")

    if failures >= 2:
        return Decision.QUARANTINE, Risk.CRITICAL
    if failures == 1:
        return Decision.QUARANTINE if mode == Mode.STRICT else Decision.REJECT, Risk.HIGH
    if warnings >= 3:
        return Decision.HOLD, Risk.HIGH
    if warnings > 0:
        return Decision.HOLD, Risk.MEDIUM
    return Decision.ACCEPT, Risk.LOW


def load_candidate(path_arg: str) -> str:
    if path_arg == "-":
        return sys.stdin.read()
    return Path(path_arg).read_text(encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description="KOTC completion simulator")
    parser.add_argument("--candidate", required=True, help="Candidate text file, or '-' for stdin")
    parser.add_argument("--target", required=True, help="Target module/path")
    parser.add_argument("--mode", choices=[m.value for m in Mode], default="REPAIR")
    parser.add_argument("--context", action="append", default=[], help="Context file path; repeatable")
    parser.add_argument("--completion-id", default="kotc_local", help="Receipt completion id suffix/name")
    args = parser.parse_args()

    mode = Mode(args.mode)
    candidate = load_candidate(args.candidate)
    checks, notes = check_policy(candidate, mode, args.target)
    decision, risk = decide(checks, mode)
    failures = sum(1 for value in checks.values() if value == "failed")
    kot_units = estimate_kot_units(candidate, mode, failures)

    completion_id = args.completion_id
    if not completion_id.startswith("kotc_"):
        completion_id = "kotc_" + completion_id

    receipt = CompletionReceipt(
        receipt_type="kotc.completion.v1",
        completion_id=completion_id,
        mode=mode.value,
        target_module=args.target,
        context_files=args.context,
        symbol_refs=extract_symbol_refs(candidate),
        kot_cost_q16=q16_hex(kot_units),
        risk=risk.value,
        decision=decision.value,
        policy_checks=checks,
        candidate_hash=sha256_text(candidate),
        notes=notes,
    )

    print(json.dumps(asdict(receipt), indent=2, sort_keys=True))
    return 0 if decision in {Decision.ACCEPT, Decision.HOLD} else 2


if __name__ == "__main__":
    raise SystemExit(main())