Context Engineering

#!/usr/bin/env python3
"""
Compression Evaluator - Evaluate compression quality with probe-based testing.

Usage:
    python compression_evaluator.py evaluate <original_file> <compressed_file>
    python compression_evaluator.py generate-probes <context_file>
"""

import argparse
import json
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class ProbeType(Enum):
    RECALL = "recall"           # Factual retention
    ARTIFACT = "artifact"       # File tracking
    CONTINUATION = "continuation"  # Task planning
    DECISION = "decision"       # Reasoning chains


@dataclass
class Probe:
    type: ProbeType
    question: str
    ground_truth: str
    context_reference: Optional[str] = None


@dataclass
class ProbeResult:
    probe: Probe
    response: str
    scores: dict
    overall_score: float


@dataclass
class EvaluationReport:
    compression_ratio: float
    quality_score: float
    dimension_scores: dict
    probe_results: list
    recommendations: list = field(default_factory=list)


# Six evaluation dimensions with weights
DIMENSIONS = {
    "accuracy": {"weight": 0.20, "description": "Technical correctness"},
    "context_awareness": {"weight": 0.15, "description": "Conversation state"},
    "artifact_trail": {"weight": 0.20, "description": "File tracking"},
    "completeness": {"weight": 0.20, "description": "Coverage and depth"},
    "continuity": {"weight": 0.15, "description": "Work continuation"},
    "instruction_following": {"weight": 0.10, "description": "Constraint adherence"}
}


def estimate_tokens(text: str) -> int:
    """Estimate token count."""
    return len(text) // 4


def extract_facts(messages: list) -> list:
    """Extract factual statements that can be probed."""
    facts = []
    patterns = [
        (r"error[:\s]+([^.]+)", "error"),
        (r"next step[s]?[:\s]+([^.]+)", "next_step"),
        (r"decided to\s+([^.]+)", "decision"),
        (r"implemented\s+([^.]+)", "implementation"),
        (r"found that\s+([^.]+)", "finding")
    ]

    for msg in messages:
        content = str(msg.get("content", "") if isinstance(msg, dict) else msg)
        for pattern, fact_type in patterns:
            matches = re.findall(pattern, content, re.IGNORECASE)
            for match in matches:
                facts.append({"type": fact_type, "content": match.strip()})
    return facts


def extract_files(messages: list) -> list:
    """Extract file references."""
    files = []
    patterns = [
        r"(?:created|modified|updated|edited|read)\s+[`'\"]?([a-zA-Z0-9_/.-]+\.[a-zA-Z]+)[`'\"]?",
        r"file[:\s]+[`'\"]?([a-zA-Z0-9_/.-]+\.[a-zA-Z]+)[`'\"]?"
    ]

    for msg in messages:
        content = str(msg.get("content", "") if isinstance(msg, dict) else msg)
        for pattern in patterns:
            matches = re.findall(pattern, content)
            files.extend(matches)
    return list(set(files))


def extract_decisions(messages: list) -> list:
    """Extract decision points."""
    decisions = []
    patterns = [
        r"chose\s+([^.]+)\s+(?:because|since|over)",
        r"decided\s+(?:to\s+)?([^.]+)",
        r"went with\s+([^.]+)"
    ]

    for msg in messages:
        content = str(msg.get("content", "") if isinstance(msg, dict) else msg)
        for pattern in patterns:
            matches = re.findall(pattern, content, re.IGNORECASE)
            decisions.extend(matches)
    return decisions


def generate_probes(messages: list) -> list:
    """Generate probe set for evaluation."""
    probes = []

    # Recall probes from facts
    facts = extract_facts(messages)
    for fact in facts[:3]:  # Limit to 3 recall probes
        probes.append(Probe(
            type=ProbeType.RECALL,
            question=f"What was the {fact['type'].replace('_', ' ')}?",
            ground_truth=fact["content"]
        ))

    # Artifact probes from files
    files = extract_files(messages)
    if files:
        probes.append(Probe(
            type=ProbeType.ARTIFACT,
            question="Which files have been modified or created?",
            ground_truth=", ".join(files)
        ))

    # Continuation probe
    probes.append(Probe(
        type=ProbeType.CONTINUATION,
        question="What should be done next?",
        ground_truth="[Extracted from context]"  # Would need LLM to generate
    ))

    # Decision probes
    decisions = extract_decisions(messages)
    for decision in decisions[:2]:  # Limit to 2 decision probes
        probes.append(Probe(
            type=ProbeType.DECISION,
            question=f"Why was the decision made to {decision[:50]}...?",
            ground_truth=decision
        ))

    return probes


def evaluate_response(probe: Probe, response: str) -> dict:
    """
    Evaluate response against probe.
    Note: Production should use LLM-as-Judge.
    """
    scores = {}
    response_lower = response.lower()
    ground_truth_lower = probe.ground_truth.lower()

    # Heuristic scoring (replace with LLM evaluation in production)
    # Check for ground truth presence
    if ground_truth_lower in response_lower:
        base_score = 1.0
    elif any(word in response_lower for word in ground_truth_lower.split()[:3]):
        base_score = 0.6
    else:
        base_score = 0.3

    # Adjust based on probe type
    if probe.type == ProbeType.ARTIFACT:
        # Check file mentions
        files_mentioned = len(re.findall(r'\.[a-z]+', response_lower))
        scores["artifact_trail"] = min(1.0, base_score + files_mentioned * 0.1)
        scores["accuracy"] = base_score
    elif probe.type == ProbeType.RECALL:
        scores["accuracy"] = base_score
        scores["completeness"] = base_score
    elif probe.type == ProbeType.CONTINUATION:
        scores["continuity"] = base_score
        scores["context_awareness"] = base_score
    elif probe.type == ProbeType.DECISION:
        scores["accuracy"] = base_score
        scores["context_awareness"] = base_score

    return scores


def calculate_compression_ratio(original: str, compressed: str) -> float:
    """Calculate compression ratio."""
    original_tokens = estimate_tokens(original)
    compressed_tokens = estimate_tokens(compressed)
    if original_tokens == 0:
        return 0.0
    return 1.0 - (compressed_tokens / original_tokens)


def evaluate_compression(original_messages: list, compressed_text: str,
                         probes: Optional[list] = None) -> EvaluationReport:
    """
    Evaluate compression quality.

    Args:
        original_messages: Original context messages
        compressed_text: Compressed summary
        probes: Optional pre-generated probes

    Returns:
        EvaluationReport with scores and recommendations
    """
    # Generate probes if not provided
    if probes is None:
        probes = generate_probes(original_messages)

    # Calculate compression ratio
    original_text = json.dumps(original_messages)
    compression_ratio = calculate_compression_ratio(original_text, compressed_text)

    # Evaluate each probe (simulated - production uses LLM)
    probe_results = []
    dimension_scores = {dim: [] for dim in DIMENSIONS}

    for probe in probes:
        # In production, send compressed_text + probe.question to LLM
        # Here we simulate with heuristic check
        scores = evaluate_response(probe, compressed_text)

        overall = sum(scores.values()) / len(scores) if scores else 0
        probe_results.append(ProbeResult(
            probe=probe,
            response="[Would be LLM response]",
            scores=scores,
            overall_score=overall
        ))

        # Aggregate by dimension
        for dim, score in scores.items():
            if dim in dimension_scores:
                dimension_scores[dim].append(score)

    # Calculate dimension averages
    avg_dimensions = {}
    for dim, scores in dimension_scores.items():
        avg_dimensions[dim] = sum(scores) / len(scores) if scores else 0.5

    # Calculate weighted quality score
    quality_score = sum(
        avg_dimensions.get(dim, 0.5) * info["weight"]
        for dim, info in DIMENSIONS.items()
    )

    # Generate recommendations
    recommendations = []
    if compression_ratio > 0.99:
        recommendations.append("Very high compression. Risk of information loss.")
    if avg_dimensions.get("artifact_trail", 1) < 0.5:
        recommendations.append("Artifact tracking weak. Add explicit file section to summary.")
    if avg_dimensions.get("continuity", 1) < 0.5:
        recommendations.append("Continuity low. Add 'Next Steps' section to summary.")
    if quality_score < 0.6:
        recommendations.append("Quality below threshold. Consider less aggressive compression.")

    return EvaluationReport(
        compression_ratio=compression_ratio,
        quality_score=quality_score,
        dimension_scores=avg_dimensions,
        probe_results=probe_results,
        recommendations=recommendations
    )


def main():
    parser = argparse.ArgumentParser(description="Compression quality evaluator")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # Evaluate command
    eval_parser = subparsers.add_parser("evaluate", help="Evaluate compression quality")
    eval_parser.add_argument("original_file", help="JSON file with original messages")
    eval_parser.add_argument("compressed_file", help="Text file with compressed summary")

    # Generate probes command
    probe_parser = subparsers.add_parser("generate-probes", help="Generate evaluation probes")
    probe_parser.add_argument("context_file", help="JSON file with context messages")

    args = parser.parse_args()

    if args.command == "evaluate":
        with open(args.original_file) as f:
            original = json.load(f)
        messages = original if isinstance(original, list) else original.get("messages", [])

        with open(args.compressed_file) as f:
            compressed = f.read()

        report = evaluate_compression(messages, compressed)
        print(json.dumps({
            "compression_ratio": f"{report.compression_ratio:.1%}",
            "quality_score": f"{report.quality_score:.2f}",
            "dimension_scores": {k: f"{v:.2f}" for k, v in report.dimension_scores.items()},
            "probe_count": len(report.probe_results),
            "recommendations": report.recommendations
        }, indent=2))

    elif args.command == "generate-probes":
        with open(args.context_file) as f:
            data = json.load(f)
        messages = data if isinstance(data, list) else data.get("messages", [])

        probes = generate_probes(messages)
        output = []
        for probe in probes:
            output.append({
                "type": probe.type.value,
                "question": probe.question,
                "ground_truth": probe.ground_truth
            })
        print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Context Analyzer - Health analysis and degradation detection for agent contexts.

Usage:
    python context_analyzer.py analyze <context_file>
    python context_analyzer.py budget --system 2000 --tools 1500 --docs 3000 --history 5000
"""

import argparse
import json
import math
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class HealthStatus(Enum):
    HEALTHY = "healthy"
    WARNING = "warning"
    DEGRADED = "degraded"
    CRITICAL = "critical"


@dataclass
class ContextAnalysis:
    total_tokens: int
    token_limit: int
    utilization: float
    health_status: HealthStatus
    health_score: float
    degradation_risk: float
    poisoning_risk: float
    recommendations: list = field(default_factory=list)


def estimate_tokens(text: str) -> int:
    """Estimate token count (~4 chars per token for English)."""
    return len(text) // 4


def estimate_message_tokens(messages: list) -> int:
    """Estimate tokens in message list."""
    total = 0
    for msg in messages:
        if isinstance(msg, dict):
            content = msg.get("content", "")
            total += estimate_tokens(str(content))
            # Add overhead for role, metadata
            total += 10
        else:
            total += estimate_tokens(str(msg))
    return total


def measure_attention_distribution(context_length: int, sample_size: int = 100) -> list:
    """
    Simulate U-shaped attention distribution.
    Real implementation would extract from model attention weights.
    """
    attention = []
    for i in range(sample_size):
        position = i / sample_size
        # U-shaped curve: high at start/end, low in middle
        if position < 0.1:
            score = 0.9 - position * 2
        elif position > 0.9:
            score = 0.7 + (position - 0.9) * 2
        else:
            score = 0.3 + 0.1 * math.sin(position * math.pi)
        attention.append(score)
    return attention


def detect_lost_in_middle(messages: list, critical_keywords: list) -> list:
    """Identify critical items in attention-degraded regions."""
    if not messages:
        return []

    total = len(messages)
    warnings = []

    for i, msg in enumerate(messages):
        position = i / total
        content = str(msg.get("content", "") if isinstance(msg, dict) else msg)

        # Middle region (10%-90%)
        if 0.1 < position < 0.9:
            for keyword in critical_keywords:
                if keyword.lower() in content.lower():
                    warnings.append({
                        "position": i,
                        "position_pct": f"{position:.1%}",
                        "keyword": keyword,
                        "risk": "high" if 0.3 < position < 0.7 else "medium"
                    })
    return warnings


def detect_poisoning_patterns(messages: list) -> dict:
    """Detect potential context poisoning indicators."""
    error_patterns = [
        r"error", r"failed", r"exception", r"cannot", r"unable",
        r"invalid", r"not found", r"undefined", r"null"
    ]
    # Simple contradiction check - look for both positive and negative statements
    contradiction_keywords = [
        ("is correct", "is not correct"),
        ("should work", "should not work"),
        ("will succeed", "will fail"),
        ("is valid", "is invalid"),
    ]

    errors_found = []
    contradictions = []

    for i, msg in enumerate(messages):
        content = str(msg.get("content", "") if isinstance(msg, dict) else msg).lower()

        # Check error patterns
        for pattern in error_patterns:
            if re.search(pattern, content):
                errors_found.append({"position": i, "pattern": pattern})

        # Check for contradiction keywords (simplified)
        for pos_phrase, neg_phrase in contradiction_keywords:
            if pos_phrase in content and neg_phrase in content:
                contradictions.append({"position": i, "type": "self-contradiction"})

    total = max(len(messages), 1)
    return {
        "error_density": len(errors_found) / total,
        "contradiction_count": len(contradictions),
        "poisoning_risk": min(1.0, (len(errors_found) * 0.1 + len(contradictions) * 0.3))
    }


def calculate_health_score(utilization: float, degradation_risk: float, poisoning_risk: float) -> float:
    """
    Calculate composite health score.
    1.0 = healthy, 0.0 = critical
    """
    score = 1.0
    # Utilization penalty (kicks in after 70%)
    if utilization > 0.7:
        score -= (utilization - 0.7) * 1.5
    # Degradation penalty
    score -= degradation_risk * 0.3
    # Poisoning penalty
    score -= poisoning_risk * 0.2
    return max(0.0, min(1.0, score))


def get_health_status(score: float) -> HealthStatus:
    """Map health score to status."""
    if score > 0.8:
        return HealthStatus.HEALTHY
    elif score > 0.6:
        return HealthStatus.WARNING
    elif score > 0.4:
        return HealthStatus.DEGRADED
    return HealthStatus.CRITICAL


def analyze_context(messages: list, token_limit: int = 128000,
                    critical_keywords: Optional[list] = None) -> ContextAnalysis:
    """
    Comprehensive context health analysis.

    Args:
        messages: List of context messages
        token_limit: Model's context window size
        critical_keywords: Keywords that should be at attention-favored positions

    Returns:
        ContextAnalysis with health metrics and recommendations
    """
    critical_keywords = critical_keywords or ["goal", "task", "important", "critical", "must"]

    # Calculate token utilization
    total_tokens = estimate_message_tokens(messages)
    utilization = total_tokens / token_limit

    # Check for lost-in-middle issues
    middle_warnings = detect_lost_in_middle(messages, critical_keywords)
    degradation_risk = min(1.0, len(middle_warnings) * 0.2)

    # Check for poisoning
    poisoning = detect_poisoning_patterns(messages)
    poisoning_risk = poisoning["poisoning_risk"]

    # Calculate health
    health_score = calculate_health_score(utilization, degradation_risk, poisoning_risk)
    health_status = get_health_status(health_score)

    # Generate recommendations
    recommendations = []
    if utilization > 0.8:
        recommendations.append("URGENT: Context utilization >80%. Trigger compaction immediately.")
    elif utilization > 0.7:
        recommendations.append("WARNING: Context utilization >70%. Plan for compaction.")

    if middle_warnings:
        recommendations.append(f"Found {len(middle_warnings)} critical items in middle region. "
                               "Consider moving to beginning/end.")

    if poisoning_risk > 0.3:
        recommendations.append("High poisoning risk detected. Review recent tool outputs for errors.")

    if health_status == HealthStatus.CRITICAL:
        recommendations.append("CRITICAL: Consider context reset with clean state.")

    return ContextAnalysis(
        total_tokens=total_tokens,
        token_limit=token_limit,
        utilization=utilization,
        health_status=health_status,
        health_score=health_score,
        degradation_risk=degradation_risk,
        poisoning_risk=poisoning_risk,
        recommendations=recommendations
    )


def calculate_budget(system: int, tools: int, docs: int, history: int,
                     buffer_pct: float = 0.15) -> dict:
    """Calculate context budget allocation."""
    subtotal = system + tools + docs + history
    buffer = int(subtotal * buffer_pct)
    total = subtotal + buffer

    return {
        "allocation": {
            "system_prompt": system,
            "tool_definitions": tools,
            "retrieved_docs": docs,
            "message_history": history,
            "reserved_buffer": buffer
        },
        "total_budget": total,
        "warning_threshold": int(total * 0.7),
        "critical_threshold": int(total * 0.8),
        "recommendations": [
            f"Trigger compaction at {int(total * 0.7):,} tokens",
            f"Aggressive optimization at {int(total * 0.8):,} tokens",
            f"Reserved {buffer:,} tokens ({buffer_pct:.0%}) for responses"
        ]
    }


def main():
    parser = argparse.ArgumentParser(description="Context health analyzer")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # Analyze command
    analyze_parser = subparsers.add_parser("analyze", help="Analyze context health")
    analyze_parser.add_argument("context_file", help="JSON file with messages array")
    analyze_parser.add_argument("--limit", type=int, default=128000, help="Token limit")
    analyze_parser.add_argument("--keywords", nargs="+", help="Critical keywords to track")

    # Budget command
    budget_parser = subparsers.add_parser("budget", help="Calculate context budget")
    budget_parser.add_argument("--system", type=int, default=2000, help="System prompt tokens")
    budget_parser.add_argument("--tools", type=int, default=1500, help="Tool definitions tokens")
    budget_parser.add_argument("--docs", type=int, default=3000, help="Retrieved docs tokens")
    budget_parser.add_argument("--history", type=int, default=5000, help="Message history tokens")
    budget_parser.add_argument("--buffer", type=float, default=0.15, help="Buffer percentage")

    args = parser.parse_args()

    if args.command == "analyze":
        with open(args.context_file) as f:
            data = json.load(f)
        messages = data if isinstance(data, list) else data.get("messages", [])
        result = analyze_context(messages, args.limit, args.keywords)
        print(json.dumps({
            "total_tokens": result.total_tokens,
            "token_limit": result.token_limit,
            "utilization": f"{result.utilization:.1%}",
            "health_status": result.health_status.value,
            "health_score": f"{result.health_score:.2f}",
            "degradation_risk": f"{result.degradation_risk:.2f}",
            "poisoning_risk": f"{result.poisoning_risk:.2f}",
            "recommendations": result.recommendations
        }, indent=2))

    elif args.command == "budget":
        result = calculate_budget(args.system, args.tools, args.docs, args.history, args.buffer)
        print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()