src/gate.py

"""
Builder Gate and Evidence Gate for CSC Engine.

Builder Gate: validates code before execution — syntax, prose detection, import check.
Evidence Gate: validates that sufficient intent exists before task-specific code generation.
"""

import ast
import sys
import pkgutil
import re
from dataclasses import dataclass, field


# ─── Installed package cache ───

_installed_packages: set[str] | None = None


def _get_installed_packages() -> set[str]:
    global _installed_packages
    if _installed_packages is None:
        _installed_packages = set()
        for m in pkgutil.iter_modules():
            _installed_packages.add(m.name)
        # Add common stdlib modules that might not show up in iter_modules
        _installed_packages.update(sys.stdlib_module_names)
        # Add common aliases (only packages actually installed)
        _installed_packages.update({"PIL", "Pillow"})
    return _installed_packages


# ─── Builder Gate ───

@dataclass
class GateResult:
    passed: bool
    reason: str
    code_extracted: str = ""
    checks: list[dict] = field(default_factory=list)


def builder_gate(code: str) -> GateResult:
    """Validate code before execution.
    
    Checks:
    1. Syntax validation — must parse as valid Python (also catches prose)
    2. Import validation — all imports must be from installed packages
    3. Noninteractive check — reject input() calls and interactive patterns
    """
    checks = []
    code = code.strip() if code else ""

    # Check 1: Empty
    if not code:
        return GateResult(passed=False, reason="No code provided", checks=[{"check": "empty", "passed": False}])

    # Check 2: Syntax validation (this also catches prose — prose won't parse as Python)
    try:
        ast.parse(code)
    except SyntaxError as e:
        # If syntax fails, check if it's prose vs actual code error
        python_indicators = [
            r'\bdef\b', r'\bclass\b', r'\bimport\b', r'\bfrom\b', r'\bif\b', r'\bfor\b',
            r'\bwhile\b', r'\breturn\b', r'\bprint\s*\(', r'\bassert\b', r'\bwith\b',
            r'\btry\b', r'\bexcept\b', r'\braise\b', r'\byield\b', r'\blambda\b',
        ]
        indicator_count = sum(1 for p in python_indicators if re.search(p, code))
        if indicator_count == 0:
            checks.append({"check": "prose_detection", "passed": False,
                           "detail": "No Python indicators found — likely prose, not code"})
            return GateResult(passed=False, reason="Code looks like prose, not executable Python",
                              checks=checks)
        checks.append({"check": "syntax", "passed": False, "detail": f"SyntaxError: {e.msg} (line {e.lineno})"})
        return GateResult(passed=False, reason=f"Syntax error: {e.msg} at line {e.lineno}",
                          checks=checks)

    checks.append({"check": "syntax", "passed": True})

    # Check 4: Import validation
    tree = ast.parse(code)
    imports = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                imports.append(alias.name.split('.')[0])
        elif isinstance(node, ast.ImportFrom):
            if node.module:
                imports.append(node.module.split('.')[0])

    installed = _get_installed_packages()
    missing = [imp for imp in imports if imp not in installed and imp != "__future__"]
    if missing:
        checks.append({"check": "imports", "passed": False,
                       "detail": f"Missing packages: {', '.join(missing)}"})
        return GateResult(passed=False, reason=f"Missing dependencies: {', '.join(missing)}. Install them or use only available packages.",
                          checks=checks)

    checks.append({"check": "imports", "passed": True, "detail": f"{len(imports)} imports validated"})

    # Check 5: Noninteractive check
    interactive_patterns = [r'\binput\s*\(', r'\braw_input\s*\(', r'\bgetpass\s*\(']
    for pattern in interactive_patterns:
        if re.search(pattern, code):
            checks.append({"check": "noninteractive", "passed": False,
                           "detail": f"Interactive call detected: {pattern}"})
            return GateResult(passed=False, reason="Code contains interactive input — not allowed in sandboxed execution",
                              checks=checks)

    checks.append({"check": "noninteractive", "passed": True})

    return GateResult(passed=True, reason="All gate checks passed", code_extracted=code, checks=checks)


# ─── Evidence Gate ───

@dataclass
class EvidenceGateResult:
    passed: bool
    reason: str
    evidence_type: str = ""
    intent_sources: list[str] = field(default_factory=list)
    fallback_level: int = 0
    artifact_type: str = ""
    sensory_channels: list[str] = field(default_factory=list)
    feature_attribution: dict = field(default_factory=dict)


def evidence_gate(observer_output: str, state_dict: dict) -> EvidenceGateResult:
    """Sensory Proprietary Compiler V1 — Fallback Ladder.
    
    NEVER returns zero artifact. Every sensory input produces something useful.
    
    Fallback levels:
    1. Explicit intent → task code
    2. Weak intent + rich sensory → instrumentation code
    3. Distinctive sensory features → aesthetic/system motifs
    4. Background audio only → topic-to-tool associations
    5. Minimal signal → capture protocol improvement
    """
    obs = (observer_output or "").lower()
    
    # Gather intent sources
    intent_sources = []
    sensory_channels = []
    feature_attribution = {}
    
    # Check for user speech / transcript
    speakers = state_dict.get("speakers", {})
    user_transcript = speakers.get("user", {}).get("transcript", "")
    if user_transcript and user_transcript.strip():
        intent_sources.append("user_speech")
        sensory_channels.append("user_voice")
        feature_attribution["user_speech"] = user_transcript[:200]
    
    # Check for any audio chunks with content
    audio_chunks = state_dict.get("audio", {}).get("chunks", [])
    has_audio_content = any(c.get("transcript", "").strip() for c in audio_chunks if isinstance(c, dict))
    if has_audio_content:
        intent_sources.append("audio_transcript")
        sensory_channels.append("audio")
    
    # Check for background speakers (TV, other people)
    for spk, info in speakers.items():
        if spk != "user" and info.get("transcript", "").strip():
            sensory_channels.append(f"background_voice:{spk}")
            feature_attribution[f"background_{spk}"] = info["transcript"][:200]
    
    # Check for visible screen text / code in observer output
    screen_indicators = ["screen", "code on", "text on", "monitor", "display", "laptop", "ide", "editor", "terminal"]
    has_screen_evidence = any(s in obs for s in screen_indicators)
    if has_screen_evidence:
        intent_sources.append("visible_screen")
        sensory_channels.append("screen")
    
    # Check for explicit typed intent / prior goal
    if "goal" in obs or "intent" in obs or "task" in obs or "want" in obs:
        intent_sources.append("explicit_intent")
    
    # Check for camera sensory features
    camera_indicators = ["face", "person", "movement", "object", "light", "wall", "fabric", "hand", "gesture",
                         "hair", "color", "purple", "brown", "breathing", "motion", "room"]
    has_camera_evidence = any(s in obs for s in camera_indicators)
    if has_camera_evidence:
        sensory_channels.append("camera")
        # Extract specific features
        for indicator in camera_indicators:
            if indicator in obs:
                feature_attribution[f"visual_{indicator}"] = True
    
    # Check motion score
    motion_score = state_dict.get("motion_score", 0.0)
    if motion_score > 0.01:
        sensory_channels.append("motion")
        feature_attribution["motion_score"] = round(motion_score, 4)
    
    # Check frame count (camera active)
    visual = state_dict.get("visual", {})
    frame_count = visual.get("frame_count", 0)
    if frame_count > 0:
        sensory_channels.append("frames")
        feature_attribution["frame_count"] = frame_count
        feature_attribution["avg_entropy"] = visual.get("avg_entropy", 0)
        feature_attribution["avg_motion"] = visual.get("avg_motion", 0)
    
    # Check audio chunk count
    audio_chunk_count = state_dict.get("audio", {}).get("chunk_count", 0)
    if audio_chunk_count > 0:
        feature_attribution["audio_chunks"] = audio_chunk_count
    
    # ─── Fallback Ladder ───
    
    # Level 1: Explicit intent → task code
    if intent_sources:
        return EvidenceGateResult(
            passed=True,
            reason=f"Level 1: Explicit intent detected. Sources: {', '.join(intent_sources)}",
            evidence_type="sufficient_intent",
            intent_sources=intent_sources,
            fallback_level=1,
            artifact_type="task_code",
            sensory_channels=sensory_channels,
            feature_attribution=feature_attribution,
        )
    
    # Level 2: Weak intent but rich sensory features → instrumentation code
    if has_camera_evidence and (len(sensory_channels) >= 2 or motion_score > 0.05):
        return EvidenceGateResult(
            passed=True,
            reason=f"Level 2: Rich sensory evidence ({', '.join(sensory_channels)}). Generating instrumentation code.",
            evidence_type="sensory_rich",
            intent_sources=[],
            fallback_level=2,
            artifact_type="instrumentation_code",
            sensory_channels=sensory_channels,
            feature_attribution=feature_attribution,
        )
    
    # Level 3: Distinctive sensory features → aesthetic/system motifs
    if has_camera_evidence or frame_count > 0:
        return EvidenceGateResult(
            passed=True,
            reason=f"Level 3: Distinctive sensory features ({', '.join(sensory_channels)}). Generating motif/design artifact.",
            evidence_type="sensory_distinctive",
            intent_sources=[],
            fallback_level=3,
            artifact_type="aesthetic_motif",
            sensory_channels=sensory_channels,
            feature_attribution=feature_attribution,
        )
    
    # Level 4: Background audio only → topic-to-tool associations
    if audio_chunk_count > 0 or any("background_voice" in ch for ch in sensory_channels):
        return EvidenceGateResult(
            passed=True,
            reason="Level 4: Audio evidence only. Generating topic-to-tool association artifact.",
            evidence_type="audio_only",
            intent_sources=[],
            fallback_level=4,
            artifact_type="topic_association",
            sensory_channels=sensory_channels,
            feature_attribution=feature_attribution,
        )
    
    # Level 5: Minimal signal → capture protocol improvement
    return EvidenceGateResult(
        passed=True,
        reason="Level 5: Minimal sensory signal. Generating capture protocol improvement artifact.",
        evidence_type="minimal_signal",
        intent_sources=[],
        fallback_level=5,
        artifact_type="capture_protocol",
        sensory_channels=sensory_channels,
        feature_attribution=feature_attribution,
    )


# ─── Gate log for audit ───

_gate_log: list[dict] = []


def log_gate_decision(gate_type: str, result, patch_hash: str = "", code_preview: str = ""):
    """Record gate decision for audit trail."""
    entry = {
        "gate_type": gate_type,
        "passed": result.passed,
        "reason": result.reason,
        "timestamp": __import__("time").time(),
        "patch_hash": patch_hash,
        "code_preview": code_preview[:200] if code_preview else "",
    }
    if hasattr(result, "checks"):
        entry["checks"] = result.checks
    if hasattr(result, "evidence_type"):
        entry["evidence_type"] = result.evidence_type
        entry["intent_sources"] = result.intent_sources
    if hasattr(result, "fallback_level"):
        entry["fallback_level"] = result.fallback_level
        entry["artifact_type"] = result.artifact_type
        entry["sensory_channels"] = result.sensory_channels
        entry["feature_attribution"] = result.feature_attribution
    _gate_log.append(entry)
    if len(_gate_log) > 100:
        _gate_log.pop(0)


def get_gate_log() -> list[dict]:
    return list(_gate_log)