Prompt Engineer Toolkit

#!/usr/bin/env python3
"""Score evaluation results from JSON test cases against expected outputs.

Supports three matching strategies: exact match, contains, and regex.
Reads a test suite JSON file where each test case defines an input,
the actual output, and expected criteria. Produces per-case and aggregate
scores aligned with the Prompt Engineer Toolkit evaluation rubric.

No external dependencies -- uses Python standard library only.
"""

import argparse
import json
import os
import re
import sys
from collections import Counter


# Default dimension weights from the SKILL.md evaluation rubric.
DEFAULT_WEIGHTS = {
    "adherence": 0.30,
    "accuracy": 0.30,
    "safety": 0.20,
    "format": 0.10,
    "relevance": 0.10,
}

PASS_THRESHOLD = 0.80
WARNING_THRESHOLD = 0.70


def check_exact_match(actual: str, expected: str, case_sensitive: bool = True) -> dict:
    """Check if actual output exactly matches expected string."""
    if case_sensitive:
        matched = actual.strip() == expected.strip()
    else:
        matched = actual.strip().lower() == expected.strip().lower()
    return {
        "method": "exact_match",
        "matched": matched,
        "score": 1.0 if matched else 0.0,
        "expected": expected[:100],
        "actual_preview": actual[:100],
    }


def check_contains(actual: str, expected_phrases: list, case_sensitive: bool = True) -> dict:
    """Check if actual output contains all expected phrases."""
    results = []
    for phrase in expected_phrases:
        if case_sensitive:
            found = phrase in actual
        else:
            found = phrase.lower() in actual.lower()
        results.append({"phrase": phrase, "found": found})

    matched_count = sum(1 for r in results if r["found"])
    total = len(expected_phrases)
    score = matched_count / total if total > 0 else 0.0

    return {
        "method": "contains",
        "matched": matched_count == total,
        "score": round(score, 3),
        "matched_count": matched_count,
        "total": total,
        "details": results,
    }


def check_not_contains(actual: str, forbidden_phrases: list, case_sensitive: bool = True) -> dict:
    """Check that actual output does NOT contain any forbidden phrases."""
    results = []
    for phrase in forbidden_phrases:
        if case_sensitive:
            found = phrase in actual
        else:
            found = phrase.lower() in actual.lower()
        results.append({"phrase": phrase, "found": found})

    violations = sum(1 for r in results if r["found"])
    total = len(forbidden_phrases)
    score = 1.0 - (violations / total) if total > 0 else 1.0

    return {
        "method": "not_contains",
        "matched": violations == 0,
        "score": round(score, 3),
        "violations": violations,
        "total": total,
        "details": results,
    }


def check_regex(actual: str, pattern: str) -> dict:
    """Check if actual output matches a regex pattern."""
    try:
        match = re.search(pattern, actual, re.DOTALL)
        matched = match is not None
        return {
            "method": "regex",
            "matched": matched,
            "score": 1.0 if matched else 0.0,
            "pattern": pattern,
            "match_text": match.group(0)[:80] if match else None,
        }
    except re.error as e:
        return {
            "method": "regex",
            "matched": False,
            "score": 0.0,
            "pattern": pattern,
            "error": str(e),
        }


def check_max_tokens(actual: str, max_tokens: int) -> dict:
    """Check if output is within token budget (approximate)."""
    word_count = len(actual.split())
    estimated_tokens = int(word_count * 1.3 + 0.5)
    within = estimated_tokens <= max_tokens

    return {
        "method": "max_tokens",
        "matched": within,
        "score": 1.0 if within else max(0.0, round(1.0 - (estimated_tokens - max_tokens) / max_tokens, 3)),
        "estimated_tokens": estimated_tokens,
        "max_tokens": max_tokens,
    }


def check_required_fields(actual: str, required_fields: list) -> dict:
    """Check if output (assumed JSON) contains required fields."""
    try:
        parsed = json.loads(actual)
    except (json.JSONDecodeError, TypeError):
        # If not JSON, check for field names as text patterns.
        found = []
        missing = []
        for field in required_fields:
            if field.lower() in actual.lower():
                found.append(field)
            else:
                missing.append(field)
        score = len(found) / len(required_fields) if required_fields else 1.0
        return {
            "method": "required_fields",
            "matched": len(missing) == 0,
            "score": round(score, 3),
            "found": found,
            "missing": missing,
            "note": "Output is not valid JSON; checked as text patterns",
        }

    def flatten_keys(obj, prefix=""):
        keys = set()
        if isinstance(obj, dict):
            for k, v in obj.items():
                full_key = f"{prefix}.{k}" if prefix else k
                keys.add(full_key)
                keys.add(k)  # Also add short name
                keys.update(flatten_keys(v, full_key))
        elif isinstance(obj, list):
            for item in obj:
                keys.update(flatten_keys(item, prefix))
        return keys

    all_keys = flatten_keys(parsed)
    found = [f for f in required_fields if f in all_keys]
    missing = [f for f in required_fields if f not in all_keys]
    score = len(found) / len(required_fields) if required_fields else 1.0

    return {
        "method": "required_fields",
        "matched": len(missing) == 0,
        "score": round(score, 3),
        "found": found,
        "missing": missing,
    }


def score_test_case(test_case: dict) -> dict:
    """Score a single test case against its expected criteria."""
    test_id = test_case.get("test_id", "unknown")
    actual = test_case.get("actual", test_case.get("output", ""))
    expected = test_case.get("expected", {})
    tags = test_case.get("tags", [])

    checks = []
    scores = []

    # Exact match
    if "exact" in expected:
        case_sensitive = expected.get("case_sensitive", True)
        result = check_exact_match(actual, expected["exact"], case_sensitive)
        checks.append(result)
        scores.append(result["score"])

    # Contains
    if "contains" in expected:
        case_sensitive = expected.get("case_sensitive", True)
        result = check_contains(actual, expected["contains"], case_sensitive)
        checks.append(result)
        scores.append(result["score"])

    # Not contains
    if "not_contains" in expected:
        case_sensitive = expected.get("case_sensitive", True)
        result = check_not_contains(actual, expected["not_contains"], case_sensitive)
        checks.append(result)
        scores.append(result["score"])

    # Regex
    if "format_regex" in expected:
        result = check_regex(actual, expected["format_regex"])
        checks.append(result)
        scores.append(result["score"])

    if "regex" in expected:
        result = check_regex(actual, expected["regex"])
        checks.append(result)
        scores.append(result["score"])

    # Max tokens
    if "max_tokens" in expected:
        result = check_max_tokens(actual, expected["max_tokens"])
        checks.append(result)
        scores.append(result["score"])

    # Required fields
    if "required_fields" in expected:
        result = check_required_fields(actual, expected["required_fields"])
        checks.append(result)
        scores.append(result["score"])

    # Compute aggregate score
    if scores:
        aggregate_score = round(sum(scores) / len(scores), 3)
    else:
        aggregate_score = 0.0

    passed = aggregate_score >= PASS_THRESHOLD
    status = "pass" if passed else "warn" if aggregate_score >= WARNING_THRESHOLD else "fail"

    return {
        "test_id": test_id,
        "score": aggregate_score,
        "status": status,
        "checks_run": len(checks),
        "checks_passed": sum(1 for c in checks if c["matched"]),
        "checks": checks,
        "tags": tags,
    }


def score_suite(test_cases: list) -> dict:
    """Score an entire test suite and produce aggregate metrics."""
    results = [score_test_case(tc) for tc in test_cases]

    total = len(results)
    if total == 0:
        return {"error": "No test cases found", "results": [], "summary": {}}

    pass_count = sum(1 for r in results if r["status"] == "pass")
    warn_count = sum(1 for r in results if r["status"] == "warn")
    fail_count = sum(1 for r in results if r["status"] == "fail")
    avg_score = round(sum(r["score"] for r in results) / total, 3)

    # Scores by tag
    tag_scores = {}
    for r in results:
        for tag in r.get("tags", []):
            tag_scores.setdefault(tag, []).append(r["score"])
    tag_averages = {tag: round(sum(s) / len(s), 3) for tag, s in tag_scores.items()}

    suite_status = "pass" if avg_score >= PASS_THRESHOLD else "warn" if avg_score >= WARNING_THRESHOLD else "fail"

    return {
        "summary": {
            "total_cases": total,
            "passed": pass_count,
            "warned": warn_count,
            "failed": fail_count,
            "pass_rate": round(pass_count / total, 3),
            "average_score": avg_score,
            "suite_status": suite_status,
            "pass_threshold": PASS_THRESHOLD,
            "warning_threshold": WARNING_THRESHOLD,
        },
        "by_tag": tag_averages,
        "results": results,
    }


def format_human(suite_result: dict) -> str:
    """Format suite results for human-readable console output."""
    lines = []
    s = suite_result["summary"]

    lines.append("Evaluation Score Report")
    lines.append("=" * 56)
    lines.append(f"  Suite Status:   {s['suite_status'].upper()}")
    lines.append(f"  Average Score:  {s['average_score']:.1%}")
    lines.append(f"  Pass Rate:      {s['pass_rate']:.1%}  ({s['passed']}/{s['total_cases']})")
    lines.append(f"  Passed: {s['passed']}  |  Warned: {s['warned']}  |  Failed: {s['failed']}")
    lines.append(f"  Thresholds:     pass >= {s['pass_threshold']}, warn >= {s['warning_threshold']}")
    lines.append("")

    # By tag
    if suite_result.get("by_tag"):
        lines.append("  Scores by Tag:")
        for tag, avg in sorted(suite_result["by_tag"].items()):
            lines.append(f"    {tag:30s} {avg:.1%}")
        lines.append("")

    # Per-case detail
    lines.append("  Per-Case Results:")
    lines.append("  " + "-" * 52)
    for r in suite_result["results"]:
        status_icon = "PASS" if r["status"] == "pass" else "WARN" if r["status"] == "warn" else "FAIL"
        tag_str = ", ".join(r["tags"][:3]) if r["tags"] else ""
        lines.append(f"  [{status_icon}] {r['test_id']:30s} {r['score']:.1%}  ({r['checks_passed']}/{r['checks_run']} checks)  {tag_str}")

        # Show failing checks
        if r["status"] != "pass":
            for check in r["checks"]:
                if not check["matched"]:
                    method = check["method"]
                    if method == "contains" and "details" in check:
                        missing = [d["phrase"] for d in check["details"] if not d["found"]]
                        lines.append(f"         {method}: missing {missing[:3]}")
                    elif method == "not_contains" and "details" in check:
                        found = [d["phrase"] for d in check["details"] if d["found"]]
                        lines.append(f"         {method}: violations {found[:3]}")
                    elif method == "required_fields" and "missing" in check:
                        lines.append(f"         {method}: missing {check['missing'][:3]}")
                    elif method == "regex":
                        lines.append(f"         {method}: pattern did not match")
                    elif method == "exact_match":
                        lines.append(f"         {method}: output does not match expected")
                    elif method == "max_tokens":
                        lines.append(f"         {method}: {check.get('estimated_tokens', '?')} tokens (max {check.get('max_tokens', '?')})")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Score evaluation results from JSON test cases against expected outputs (exact match, contains, regex).",
        epilog=(
            "Test case JSON format:\n"
            '  [{"test_id": "tc-001", "actual": "output text",\n'
            '    "expected": {"contains": ["word1"], "regex": "^pattern$"},\n'
            '    "tags": ["happy-path"]}]\n\n'
            "Example: python eval_scorer.py test_results.json --json"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("suite_file", help="Path to JSON file containing test cases with actual outputs and expected criteria")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    parser.add_argument("--fail-under", type=float, default=None,
                        help="Exit with code 1 if average score is below this threshold (e.g., 0.80)")
    args = parser.parse_args()

    if not os.path.isfile(args.suite_file):
        print(f"Error: file not found: {args.suite_file}", file=sys.stderr)
        sys.exit(1)

    with open(args.suite_file, "r", encoding="utf-8") as f:
        try:
            data = json.load(f)
        except json.JSONDecodeError as e:
            print(f"Error: invalid JSON in {args.suite_file}: {e}", file=sys.stderr)
            sys.exit(1)

    # Accept either a list or a dict with a "test_cases" key.
    if isinstance(data, dict):
        test_cases = data.get("test_cases", data.get("tests", []))
    elif isinstance(data, list):
        test_cases = data
    else:
        print("Error: JSON root must be a list of test cases or an object with a 'test_cases' key.", file=sys.stderr)
        sys.exit(1)

    if not test_cases:
        print("Error: no test cases found in input file.", file=sys.stderr)
        sys.exit(1)

    suite_result = score_suite(test_cases)

    if args.json:
        print(json.dumps(suite_result, indent=2))
    else:
        print(format_human(suite_result))

    # Exit code for CI integration.
    if args.fail_under is not None:
        avg = suite_result["summary"]["average_score"]
        if avg < args.fail_under:
            sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Analyze prompt files for quality metrics.

Evaluates prompts against production quality criteria including clarity score,
instruction density, few-shot coverage, token estimation, and structural layer
detection based on the 6-layer system prompt architecture.

No external dependencies -- uses Python standard library only.
"""

import argparse
import json
import math
import os
import re
import sys
from collections import Counter


# Approximate token count using whitespace + punctuation splitting.
# This is a rough heuristic (~1.3 tokens per whitespace-delimited word for English).
TOKEN_RATIO = 1.3

# Instruction signal words that indicate directive content.
INSTRUCTION_SIGNALS = [
    r"\bmust\b", r"\bshould\b", r"\bshall\b", r"\balways\b", r"\bnever\b",
    r"\bdo not\b", r"\bdon't\b", r"\bavoid\b", r"\bensure\b", r"\brequire",
    r"\bcritical\b", r"\bimportant\b", r"\brule\b", r"\bconstraint\b",
    r"\bprohibit", r"\bforbid", r"\bmandatory\b", r"\bonly\b",
]

# Few-shot indicator patterns.
FEW_SHOT_PATTERNS = [
    r"(?i)\bexample\s*\d*\s*[:(\[]",
    r"(?i)\binput\s*:",
    r"(?i)\boutput\s*:",
    r"(?i)\bsample\b",
    r"(?i)\bdemonstrat",
    r"(?i)\bfor instance\b",
    r"(?i)\be\.g\.\b",
]

# Layer detection heuristics aligned with the 6-layer architecture.
LAYER_PATTERNS = {
    "identity_role": [
        r"(?i)you are\b", r"(?i)act as\b", r"(?i)your role\b",
        r"(?i)you serve as\b", r"(?i)as a .{3,30} expert",
    ],
    "capabilities_constraints": [
        r"(?i)you can\b", r"(?i)you cannot\b", r"(?i)you have access\b",
        r"(?i)you are able\b", r"(?i)capable of\b", r"(?i)limited to\b",
    ],
    "output_format": [
        r"(?i)respond with\b", r"(?i)output format\b", r"(?i)json\b",
        r"(?i)structure your\b", r"(?i)format your\b", r"(?i)schema\b",
    ],
    "quality_standards": [
        r"(?i)quality\b", r"(?i)thorough\b", r"(?i)accurate\b",
        r"(?i)concise\b", r"(?i)include .{0,20}(detail|evidence|citation)",
    ],
    "anti_patterns": [
        r"(?i)never\b", r"(?i)do not\b", r"(?i)don't\b", r"(?i)avoid\b",
        r"(?i)forbidden\b", r"(?i)prohibited\b",
    ],
    "examples": [
        r"(?i)example\b", r"(?i)for instance\b", r"(?i)here is\b",
        r"(?i)sample\b", r"(?i)demonstration\b",
    ],
}

# Clarity deductors -- patterns that reduce clarity.
CLARITY_DEDUCTORS = [
    (r"(?i)\betc\.?\b", 0.03, "vague_etc"),
    (r"(?i)\bstuff\b", 0.04, "informal_stuff"),
    (r"(?i)\bthings\b", 0.02, "vague_things"),
    (r"(?i)\bmaybe\b", 0.03, "hedging_maybe"),
    (r"(?i)\bsomehow\b", 0.04, "vague_somehow"),
    (r"(?i)\bkind of\b", 0.03, "hedging_kind_of"),
    (r"(?i)\bsort of\b", 0.03, "hedging_sort_of"),
    (r"(?i)\bprobably\b", 0.02, "hedging_probably"),
    (r"(?i)\bbasically\b", 0.02, "filler_basically"),
    (r"(?i)\bjust\b", 0.01, "filler_just"),
]


def estimate_tokens(text: str) -> int:
    """Rough token estimate based on word count and punctuation."""
    words = text.split()
    return max(1, int(math.ceil(len(words) * TOKEN_RATIO)))


def compute_instruction_density(text: str) -> dict:
    """Return instruction density and matched signals."""
    sentences = re.split(r"[.!?\n]", text)
    sentences = [s.strip() for s in sentences if s.strip()]
    if not sentences:
        return {"density": 0.0, "instruction_count": 0, "sentence_count": 0, "signals": []}

    instruction_sentences = 0
    matched_signals = []
    for sent in sentences:
        for pattern in INSTRUCTION_SIGNALS:
            if re.search(pattern, sent, re.IGNORECASE):
                instruction_sentences += 1
                matched_signals.append(pattern.strip(r"\b"))
                break

    density = instruction_sentences / len(sentences)
    return {
        "density": round(density, 3),
        "instruction_count": instruction_sentences,
        "sentence_count": len(sentences),
        "signals": list(set(matched_signals)),
    }


def detect_few_shot(text: str) -> dict:
    """Detect few-shot examples and estimate count."""
    example_blocks = re.findall(r"(?i)example\s*\d*\s*[:(]", text)
    input_output_pairs = min(
        len(re.findall(r"(?i)\binput\s*:", text)),
        len(re.findall(r"(?i)\boutput\s*:", text)),
    )
    pattern_hits = sum(1 for p in FEW_SHOT_PATTERNS if re.search(p, text))
    estimated_shots = max(len(example_blocks), input_output_pairs)

    coverage = "none"
    if estimated_shots >= 3:
        coverage = "good"
    elif estimated_shots >= 1:
        coverage = "partial"

    return {
        "estimated_shots": estimated_shots,
        "coverage": coverage,
        "pattern_hits": pattern_hits,
    }


def detect_layers(text: str) -> dict:
    """Detect which of the 6 architecture layers are present."""
    results = {}
    for layer, patterns in LAYER_PATTERNS.items():
        hits = sum(1 for p in patterns if re.search(p, text))
        results[layer] = {
            "detected": hits > 0,
            "signal_strength": min(hits, len(patterns)),
            "max_signals": len(patterns),
        }
    detected_count = sum(1 for v in results.values() if v["detected"])
    return {"layers": results, "detected_count": detected_count, "total_layers": 6}


def compute_clarity_score(text: str, instruction_info: dict, layer_info: dict, few_shot_info: dict) -> dict:
    """Compute an overall clarity score from 0 to 100."""
    # Start at 60 base score.
    score = 60.0
    deductions = []

    # Reward instruction density (up to +15).
    density = instruction_info["density"]
    if density >= 0.3:
        score += 15
    elif density >= 0.15:
        score += 10
    elif density >= 0.05:
        score += 5

    # Reward layer coverage (up to +15).
    layer_ratio = layer_info["detected_count"] / layer_info["total_layers"]
    score += layer_ratio * 15

    # Reward few-shot examples (up to +10).
    if few_shot_info["coverage"] == "good":
        score += 10
    elif few_shot_info["coverage"] == "partial":
        score += 5

    # Deduct for clarity anti-patterns.
    for pattern, penalty, label in CLARITY_DEDUCTORS:
        count = len(re.findall(pattern, text))
        if count > 0:
            deduction = min(penalty * count, 0.10)  # Cap per-pattern.
            score -= deduction * 100
            deductions.append({"pattern": label, "count": count, "penalty": round(deduction * 100, 1)})

    score = max(0, min(100, round(score, 1)))

    grade = "excellent" if score >= 90 else "good" if score >= 75 else "fair" if score >= 60 else "poor"

    return {"score": score, "grade": grade, "deductions": deductions}


def analyze_prompt(text: str, filepath: str = "<stdin>") -> dict:
    """Run full analysis on a prompt text and return results dict."""
    token_count = estimate_tokens(text)
    line_count = text.count("\n") + 1
    char_count = len(text)
    word_count = len(text.split())

    instruction_info = compute_instruction_density(text)
    few_shot_info = detect_few_shot(text)
    layer_info = detect_layers(text)
    clarity_info = compute_clarity_score(text, instruction_info, layer_info, few_shot_info)

    return {
        "file": filepath,
        "metrics": {
            "token_estimate": token_count,
            "line_count": line_count,
            "char_count": char_count,
            "word_count": word_count,
        },
        "clarity": clarity_info,
        "instruction_density": instruction_info,
        "few_shot": few_shot_info,
        "layer_coverage": layer_info,
    }


def format_human(result: dict) -> str:
    """Format analysis result for human-readable console output."""
    lines = []
    lines.append(f"Prompt Analysis: {result['file']}")
    lines.append("=" * 60)

    m = result["metrics"]
    lines.append(f"  Tokens (est):  {m['token_estimate']}")
    lines.append(f"  Lines:         {m['line_count']}")
    lines.append(f"  Words:         {m['word_count']}")
    lines.append(f"  Characters:    {m['char_count']}")
    lines.append("")

    c = result["clarity"]
    lines.append(f"  Clarity Score: {c['score']} / 100  ({c['grade']})")
    if c["deductions"]:
        lines.append("  Deductions:")
        for d in c["deductions"]:
            lines.append(f"    - {d['pattern']}: {d['count']} occurrence(s), -{d['penalty']} pts")
    lines.append("")

    i = result["instruction_density"]
    lines.append(f"  Instruction Density: {i['density']} ({i['instruction_count']}/{i['sentence_count']} sentences)")
    if i["signals"]:
        lines.append(f"  Signal words: {', '.join(i['signals'][:10])}")
    lines.append("")

    fs = result["few_shot"]
    lines.append(f"  Few-Shot Coverage: {fs['coverage']} ({fs['estimated_shots']} example(s) detected)")
    lines.append("")

    lc = result["layer_coverage"]
    lines.append(f"  Layer Coverage: {lc['detected_count']}/{lc['total_layers']} layers detected")
    for name, info in lc["layers"].items():
        status = "YES" if info["detected"] else " - "
        lines.append(f"    [{status}] {name.replace('_', ' ').title()}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze prompt files for quality metrics (clarity, instruction density, few-shot coverage, tokens).",
        epilog="Example: python prompt_analyzer.py my_prompt.txt --json",
    )
    parser.add_argument("files", nargs="+", help="Prompt file(s) to analyze")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    results = []
    for filepath in args.files:
        if not os.path.isfile(filepath):
            print(f"Error: file not found: {filepath}", file=sys.stderr)
            sys.exit(1)
        with open(filepath, "r", encoding="utf-8") as f:
            text = f.read()
        results.append(analyze_prompt(text, filepath))

    if args.json:
        print(json.dumps(results if len(results) > 1 else results[0], indent=2))
    else:
        for result in results:
            print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Compare two prompt versions and highlight structural changes.

Detects added/removed sections, instruction changes, layer modifications,
token count deltas, and flags potential regressions aligned with the
Prompt Diff Analysis checklist from the Prompt Engineer Toolkit.

No external dependencies -- uses Python standard library only.
"""

import argparse
import difflib
import json
import os
import re
import sys
from collections import OrderedDict


# Section header patterns (markdown-style and separator-based).
SECTION_PATTERNS = [
    re.compile(r"^(#{1,6})\s+(.+)$"),           # Markdown headers
    re.compile(r"^([A-Z][A-Za-z\s]{2,50}):$"),   # "Section Name:"
    re.compile(r"^-{3,}$"),                       # Horizontal rules
    re.compile(r"^={3,}$"),                       # Double rules
]

# Risk checklist from the SKILL.md Prompt Diff Analysis section.
RISK_CHECKS = [
    {
        "id": "constraints_removed",
        "label": "Constraints removed",
        "risk": "Safety regression",
        "patterns_old": [r"(?i)\bmust\b", r"(?i)\bnever\b", r"(?i)\bdo not\b", r"(?i)\bdon't\b",
                         r"(?i)\bshall not\b", r"(?i)\bprohibit", r"(?i)\bforbid"],
    },
    {
        "id": "examples_changed",
        "label": "Examples changed",
        "risk": "Calibration shift",
        "patterns_old": [r"(?i)\bexample\b", r"(?i)\bfor instance\b", r"(?i)\bsample\b"],
    },
    {
        "id": "output_format_changed",
        "label": "Output format changed",
        "risk": "Downstream parser breakage",
        "patterns_old": [r"(?i)\bjson\b", r"(?i)\bschema\b", r"(?i)\bformat\b",
                         r"(?i)\brespond with\b", r"(?i)\bstructure\b"],
    },
    {
        "id": "anti_patterns_removed",
        "label": "Anti-patterns removed",
        "risk": "Known failure modes may return",
        "patterns_old": [r"(?i)\bavoid\b", r"(?i)\bnever\b", r"(?i)\bdo not\b",
                         r"(?i)\bdon't\b", r"(?i)\bforbid"],
    },
]

TOKEN_RATIO = 1.3


def estimate_tokens(text: str) -> int:
    words = text.split()
    return max(1, int(len(words) * TOKEN_RATIO + 0.5))


def extract_sections(text: str) -> OrderedDict:
    """Split text into named sections based on header patterns."""
    sections = OrderedDict()
    current_name = "__preamble__"
    current_lines = []

    for line in text.splitlines():
        matched = False
        for pattern in SECTION_PATTERNS[:2]:  # Only named headers
            m = pattern.match(line.strip())
            if m:
                if current_lines or current_name != "__preamble__":
                    sections[current_name] = "\n".join(current_lines)
                current_name = m.group(2) if m.lastindex >= 2 else m.group(1)
                current_name = current_name.strip()
                current_lines = []
                matched = True
                break
        if not matched:
            current_lines.append(line)

    sections[current_name] = "\n".join(current_lines)
    return sections


def compute_line_diff(old_text: str, new_text: str) -> dict:
    """Compute unified diff statistics."""
    old_lines = old_text.splitlines(keepends=True)
    new_lines = new_text.splitlines(keepends=True)

    diff = list(difflib.unified_diff(old_lines, new_lines, fromfile="old", tofile="new", n=3))
    added = sum(1 for l in diff if l.startswith("+") and not l.startswith("+++"))
    removed = sum(1 for l in diff if l.startswith("-") and not l.startswith("---"))

    similarity = difflib.SequenceMatcher(None, old_text, new_text).ratio()

    return {
        "added_lines": added,
        "removed_lines": removed,
        "similarity": round(similarity, 3),
        "diff_text": "".join(diff),
    }


def compare_sections(old_sections: OrderedDict, new_sections: OrderedDict) -> dict:
    """Compare sections between old and new prompts."""
    old_names = set(old_sections.keys())
    new_names = set(new_sections.keys())

    added = sorted(new_names - old_names)
    removed = sorted(old_names - new_names)
    common = sorted(old_names & new_names)

    modified = []
    unchanged = []
    for name in common:
        if old_sections[name].strip() != new_sections[name].strip():
            sim = difflib.SequenceMatcher(None, old_sections[name], new_sections[name]).ratio()
            modified.append({"section": name, "similarity": round(sim, 3)})
        else:
            unchanged.append(name)

    return {
        "added": added,
        "removed": removed,
        "modified": modified,
        "unchanged": unchanged,
    }


def check_risks(old_text: str, new_text: str, line_diff: dict) -> list:
    """Run the risk checklist from the Prompt Diff Analysis."""
    findings = []

    for check in RISK_CHECKS:
        old_hits = 0
        new_hits = 0
        for pat in check["patterns_old"]:
            old_hits += len(re.findall(pat, old_text))
            new_hits += len(re.findall(pat, new_text))

        if old_hits > new_hits:
            findings.append({
                "check": check["id"],
                "label": check["label"],
                "risk": check["risk"],
                "severity": "high" if (old_hits - new_hits) > 2 else "medium",
                "detail": f"Signal count dropped from {old_hits} to {new_hits}",
            })
        elif new_hits > old_hits and check["id"] in ("output_format_changed",):
            findings.append({
                "check": check["id"],
                "label": check["label"],
                "risk": check["risk"],
                "severity": "medium",
                "detail": f"Signal count changed from {old_hits} to {new_hits}",
            })

    # Check if prompt got significantly longer (context budget risk).
    old_tokens = estimate_tokens(old_text)
    new_tokens = estimate_tokens(new_text)
    if new_tokens > old_tokens * 1.25:
        findings.append({
            "check": "prompt_length_increase",
            "label": "Prompt significantly longer",
            "risk": "Context budget impact",
            "severity": "low",
            "detail": f"Token estimate grew from {old_tokens} to {new_tokens} (+{new_tokens - old_tokens})",
        })

    return findings


def compute_instruction_delta(old_text: str, new_text: str) -> dict:
    """Compare instruction-bearing lines between versions."""
    instruction_pat = re.compile(
        r"(?i)\b(must|should|shall|always|never|do not|don't|avoid|ensure|require|critical|important)\b"
    )
    old_instructions = [l.strip() for l in old_text.splitlines() if instruction_pat.search(l)]
    new_instructions = [l.strip() for l in new_text.splitlines() if instruction_pat.search(l)]

    old_set = set(old_instructions)
    new_set = set(new_instructions)

    return {
        "old_count": len(old_instructions),
        "new_count": len(new_instructions),
        "added_instructions": sorted(new_set - old_set)[:15],
        "removed_instructions": sorted(old_set - new_set)[:15],
    }


def diff_prompts(old_path: str, new_path: str) -> dict:
    """Run full diff analysis between two prompt files."""
    with open(old_path, "r", encoding="utf-8") as f:
        old_text = f.read()
    with open(new_path, "r", encoding="utf-8") as f:
        new_text = f.read()

    old_sections = extract_sections(old_text)
    new_sections = extract_sections(new_text)

    line_diff = compute_line_diff(old_text, new_text)
    section_diff = compare_sections(old_sections, new_sections)
    risks = check_risks(old_text, new_text, line_diff)
    instruction_delta = compute_instruction_delta(old_text, new_text)

    old_tokens = estimate_tokens(old_text)
    new_tokens = estimate_tokens(new_text)

    return {
        "old_file": old_path,
        "new_file": new_path,
        "summary": {
            "old_tokens": old_tokens,
            "new_tokens": new_tokens,
            "token_delta": new_tokens - old_tokens,
            "similarity": line_diff["similarity"],
            "lines_added": line_diff["added_lines"],
            "lines_removed": line_diff["removed_lines"],
        },
        "sections": section_diff,
        "instructions": instruction_delta,
        "risks": risks,
        "risk_count": len(risks),
        "diff_text": line_diff["diff_text"],
    }


def format_human(result: dict, show_diff: bool = False) -> str:
    """Format diff result for human-readable console output."""
    lines = []
    lines.append(f"Prompt Diff: {result['old_file']} -> {result['new_file']}")
    lines.append("=" * 64)

    s = result["summary"]
    lines.append(f"  Similarity:    {s['similarity']:.1%}")
    lines.append(f"  Tokens:        {s['old_tokens']} -> {s['new_tokens']} ({'+' if s['token_delta'] >= 0 else ''}{s['token_delta']})")
    lines.append(f"  Lines added:   {s['lines_added']}")
    lines.append(f"  Lines removed: {s['lines_removed']}")
    lines.append("")

    # Sections
    sec = result["sections"]
    if sec["added"]:
        lines.append("  Sections ADDED:")
        for name in sec["added"]:
            lines.append(f"    + {name}")
    if sec["removed"]:
        lines.append("  Sections REMOVED:")
        for name in sec["removed"]:
            lines.append(f"    - {name}")
    if sec["modified"]:
        lines.append("  Sections MODIFIED:")
        for m in sec["modified"]:
            lines.append(f"    ~ {m['section']} (similarity: {m['similarity']:.1%})")
    if sec["unchanged"]:
        lines.append(f"  Sections unchanged: {len(sec['unchanged'])}")
    lines.append("")

    # Instructions
    inst = result["instructions"]
    lines.append(f"  Instructions: {inst['old_count']} -> {inst['new_count']}")
    if inst["removed_instructions"]:
        lines.append("  REMOVED instructions:")
        for i in inst["removed_instructions"][:8]:
            lines.append(f"    - {i[:80]}")
    if inst["added_instructions"]:
        lines.append("  ADDED instructions:")
        for i in inst["added_instructions"][:8]:
            lines.append(f"    + {i[:80]}")
    lines.append("")

    # Risks
    if result["risks"]:
        lines.append(f"  RISK FINDINGS ({result['risk_count']}):")
        for r in result["risks"]:
            sev_marker = "!!!" if r["severity"] == "high" else "! " if r["severity"] == "medium" else "  "
            lines.append(f"    [{sev_marker}] {r['label']}: {r['risk']}")
            lines.append(f"         {r['detail']}")
    else:
        lines.append("  No risk findings detected.")
    lines.append("")

    if show_diff and result["diff_text"]:
        lines.append("  Unified Diff:")
        lines.append("  " + "-" * 40)
        for dl in result["diff_text"].splitlines()[:60]:
            lines.append(f"  {dl}")
        if result["diff_text"].count("\n") > 60:
            lines.append("  ... (truncated)")
    lines.append("")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Compare two prompt versions and highlight structural changes, instruction deltas, and risk findings.",
        epilog="Example: python prompt_diff.py prompt_v1.txt prompt_v2.txt --show-diff",
    )
    parser.add_argument("old_file", help="Path to the old/baseline prompt file")
    parser.add_argument("new_file", help="Path to the new/candidate prompt file")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    parser.add_argument("--show-diff", action="store_true", help="Include unified diff text in output")
    args = parser.parse_args()

    for path in (args.old_file, args.new_file):
        if not os.path.isfile(path):
            print(f"Error: file not found: {path}", file=sys.stderr)
            sys.exit(1)

    result = diff_prompts(args.old_file, args.new_file)

    if args.json:
        output = result
        if not args.show_diff:
            output = {k: v for k, v in result.items() if k != "diff_text"}
        print(json.dumps(output, indent=2))
    else:
        print(format_human(result, show_diff=args.show_diff))


if __name__ == "__main__":
    main()