Initial public upload from Projekte/Sidecar-Workspace (part 3)

7777e71 verified about 1 month ago

22.6 kB

#!/usr/bin/env python3
"""Sidecar Eval Runner — replays scenario events through v1/v2/v3/v4/v5.
Usage:
    python eval_runner.py                          # run all scenarios
    python eval_runner.py loop_detection           # run single scenario by name
    python eval_runner.py --list                   # list available scenarios
    python eval_runner.py --baseline results/X.json       # save current run as baseline
    python eval_runner.py --compare results/X.json        # compare against saved baseline
    python eval_runner.py --coverage                      # show detector coverage map
"""

import copy
import glob
import importlib.util
import json
import os
import sys
import time
from pathlib import Path
from unittest.mock import MagicMock, patch

VERSIONS_DIR = Path(__file__).parent / "versions"
SCENARIOS_DIR = Path(__file__).parent / "scenarios"
RESULTS_DIR = Path(__file__).parent / "results"

RESULTS_DIR.mkdir(exist_ok=True)

# In-memory session state store — keyed by (version_name, session_id)
_SESSION_STORE: dict = {}


# ---------------------------------------------------------------------------
# Version loading
# ---------------------------------------------------------------------------

def load_version(version_name: str):
    """Import sidecar module from versions/{version_name}/ with all dependencies mocked."""
    version_dir = VERSIONS_DIR / version_name
    sidecar_path = version_dir / "sidecar.py"

    if not sidecar_path.exists():
        raise FileNotFoundError(f"No sidecar.py in {version_dir}")

    # Add version dir to front of sys.path so relative imports work
    version_str = str(version_dir)
    if version_str not in sys.path:
        sys.path.insert(0, version_str)

    # Module name must be unique per version to avoid caching collisions
    mod_name = f"sidecar_{version_name}"

    # Remove cached version if present (allows re-import with fresh state)
    for key in list(sys.modules.keys()):
        if key.startswith(f"sidecar_{version_name}") or (
            version_str in sys.modules.get(key, object).__spec__.origin
            if hasattr(sys.modules.get(key, None), "__spec__") and
               sys.modules[key].__spec__ is not None and
               sys.modules[key].__spec__.origin
            else False
        ):
            del sys.modules[key]

    spec = importlib.util.spec_from_file_location(mod_name, sidecar_path)
    module = importlib.util.module_from_spec(spec)

    # Patch companion file reads to return empty strings (no actual files needed)
    mock_open = _make_mock_open()

    patches = [
        # Companion files — mock open() in the sidecar module's builtins
        patch("builtins.open", mock_open),
        # fcntl.flock — no-op (no file locking in eval)
        patch("fcntl.flock", return_value=None),
        # tempfile.mkstemp — return a fake fd/path pair (save goes through mock open)
        patch("tempfile.mkstemp", side_effect=_fake_mkstemp),
        # os.replace — no-op (we use in-memory state)
        patch("os.replace", return_value=None),
        # os.unlink — no-op
        patch("os.unlink", return_value=None),
        # os.fdopen — return mock file object
        patch("os.fdopen", side_effect=_fake_fdopen),
    ]

    # For v2/v3/v4/v5: mock SessionWatcher so it doesn't start threads
    if version_name in ("v2", "v3", "v4", "v5"):
        patches.append(patch.dict(sys.modules, {
            "watchdog": MagicMock(),
            "watchdog.observers": MagicMock(),
            "watchdog.events": MagicMock(),
        }))

    for p in patches:
        p.start()

    try:
        spec.loader.exec_module(module)
    finally:
        for p in patches:
            try:
                p.stop()
            except RuntimeError:
                pass  # already stopped

    # Tag module so we can reference its version
    module._eval_version = version_name
    module._eval_version_dir = version_dir

    return module


def _make_mock_open():
    """Return a mock open() that returns empty content for companion files,
    and delegates lock files / tmp files to a no-op in-memory implementation."""
    real_open = open

    class _FakeFile:
        def __init__(self):
            self._data = ""
        def read(self):
            return self._data
        def write(self, data):
            pass
        def __enter__(self):
            return self
        def __exit__(self, *a):
            pass
        def fileno(self):
            return -1

    def _mock_open(path, mode="r", *args, **kwargs):
        path_str = str(path) if not isinstance(path, str) else path
        # Companion files and coaching rules — return empty
        if any(name in path_str for name in [
            "WelcheFehlerVermeiden", "Skilluebersicht", "coaching_rules",
            "minimax-key", "gemini-key", "sidecar-rules.json",
        ]):
            return _FakeFile()
        # Lock files and state files in /tmp — return no-op file
        if path_str.startswith("/tmp/sidecar"):
            return _FakeFile()
        # Log file — no-op
        if "claude-sidecar.log" in path_str:
            return _FakeFile()
        # For actual eval/version source files, use real open
        return real_open(path, mode, *args, **kwargs)

    return _mock_open


_fake_fd_counter = [100]

def _fake_mkstemp(**kwargs):
    fd = _fake_fd_counter[0]
    _fake_fd_counter[0] += 1
    return (fd, f"/tmp/fake-sidecar-{fd}.tmp")


def _fake_fdopen(fd, mode="r", *args, **kwargs):
    class _NoopFile:
        def write(self, d): pass
        def __enter__(self): return self
        def __exit__(self, *a): pass
    return _NoopFile()


# ---------------------------------------------------------------------------
# State creation with in-memory session I/O patches
# ---------------------------------------------------------------------------

def create_fresh_state(version_name: str, module):
    """Create a fresh SidecarState with file I/O replaced by in-memory dicts."""

    # Per-version session store — isolated from other versions
    store: dict = {}

    def _load_session(session_id: str) -> dict:
        if session_id not in store:
            store[session_id] = {
                "history": [],
                "last_analysis_ts": 0,
                "session_start_ts": 0,
                "files_touched": [],
                "total_calls": 0,
                "recap_done": False,
            }
        return copy.deepcopy(store[session_id])

    def _save_session(session_id: str, state: dict) -> None:
        store[session_id] = copy.deepcopy(state)

    def _lock_path_noop(session_id: str) -> str:
        return f"/tmp/eval-lock-{session_id}.lock"

    # Patch module-level functions
    module._load_session_state = _load_session
    module._save_session_state = _save_session
    module._lock_path = _lock_path_noop

    # Build state object — companion file loads fail gracefully (mock open returns "")
    mock_open = _make_mock_open()
    with patch("builtins.open", mock_open), \
         patch("fcntl.flock", return_value=None):
        state = module.SidecarState()
        # Load companion files (returns empty — no actual files needed)
        state.load_companion_files()

    # v3: disable LLM judge — no external calls
    if hasattr(state, "llm_judge"):
        state.llm_judge.enabled = False

    # v2/v3: disable SessionWatcher — stop background threads if started
    if hasattr(state, "session_watcher"):
        try:
            state.session_watcher.stop()
        except Exception:
            pass
        # Replace with a no-op stub
        state.session_watcher = MagicMock()

    return state, store


# ---------------------------------------------------------------------------
# Event processing
# ---------------------------------------------------------------------------

def process_event_for_version(module, event: dict, state) -> dict:
    """Run _process_event on a version, return response + latency."""
    full_event = {
        "session_id": event.get("session_id", "eval-session-001"),
        "tool_name": event["tool"],
        "tool_input": event.get("input", {}),
        "error": "Error occurred" if event.get("had_error") else "",
        "hook_event_name": event.get("hook_event", "PostToolUse"),
        "tool_response": event.get("response_text", ""),
    }

    # Disable cooldown in eval so all detections are captured regardless of timing
    version_mod_name = f"sidecar_{getattr(module, '_eval_version', 'vX')}"
    with patch("fcntl.flock", return_value=None), \
         patch("builtins.open", _make_mock_open()), \
         patch("tempfile.mkstemp", side_effect=_fake_mkstemp), \
         patch("os.replace", return_value=None), \
         patch("os.fdopen", side_effect=_fake_fdopen), \
         patch.object(module, "CONCERN_COOLDOWN_SECONDS", 0):
        start = time.perf_counter()
        result = module._process_event(full_event, state)
        elapsed = time.perf_counter() - start

    return {
        "response": result,
        "latency_ms": round(elapsed * 1000, 2),
        "fired": _extract_fired_detectors(result),
    }


def _extract_fired_detectors(response: dict) -> list:
    """Parse which detectors fired from the response dict."""
    fired = []
    if not response:
        return fired

    text = response.get("additionalContext", "") + response.get("systemMessage", "")
    reason = response.get("reason", "")
    all_text = text + " " + reason

    detector_names = [
        "LOOP", "READ-STORM", "ERROR-CASCADE", "DRIFT", "YOLO",
        "THRASH", "STALL", "ANTI-PATTERN", "SKILL-SUGGEST",
        "ENFORCEMENT", "SIDECAR GATE",
    ]
    for det in detector_names:
        if det in all_text:
            fired.append(det)

    if response.get("decision") == "block":
        fired.append("GATE-BLOCK")
        # Extract gate rule name from reason: "[SIDECAR GATE] gate_name: message"
        if "GATE]" in reason:
            gate_part = reason.split("GATE]")[-1].strip()
            gate_name = gate_part.split(":")[0].strip()
            # Map gate names to expected detection names
            gate_map = {
                "gate_autonomous_git": "autonomous_git",
                "gate_force_push": "force_push",
                "gate_rm_rf": "rm_rf",
                "gate_secret_exposure": "secret_exposure",
            }
            mapped = gate_map.get(gate_name, gate_name)
            if mapped not in fired:
                fired.append(mapped)

    # Extract enforcement rule names: "ENFORCEMENT [rule_name]:"
    import re
    for m in re.finditer(r"ENFORCEMENT \[(\w+)\]", all_text):
        rule_name = m.group(1)
        if rule_name not in fired:
            fired.append(rule_name)

    return fired


# ---------------------------------------------------------------------------
# Scenario runner
# ---------------------------------------------------------------------------

def run_scenario(scenario_path: Path, versions: dict) -> dict:
    """Run one scenario through all versions, return results dict."""
    with open(scenario_path) as f:
        scenario = json.load(f)

    results = {
        "scenario": scenario["name"],
        "description": scenario.get("description", ""),
        "expected_detections": scenario.get("expected_detections", []),
        "events": [],
        "summary": {},
    }

    # Create fresh state per version (isolated stores)
    states = {}
    for vname, vmod in versions.items():
        state_obj, _ = create_fresh_state(vname, vmod)
        states[vname] = (vmod, state_obj)

    for i, event in enumerate(scenario["events"]):
        event_result = {
            "index": i,
            "tool": event["tool"],
            "input": event.get("input", {}),
            "had_error": event.get("had_error", False),
            "responses": {},
        }
        for vname, (vmod, state) in states.items():
            event_result["responses"][vname] = process_event_for_version(vmod, event, state)

        results["events"].append(event_result)

    # Build per-version summary
    for vname in versions:
        all_fired = []
        total_latency = 0.0
        injections = 0
        for ev in results["events"]:
            r = ev["responses"][vname]
            all_fired.extend(r["fired"])
            total_latency += r["latency_ms"]
            if r["response"]:
                injections += 1

        expected = set(results["expected_detections"])
        found = set(all_fired)
        results["summary"][vname] = {
            "detectors_fired": list(dict.fromkeys(all_fired)),  # deduplicated, ordered
            "injections": injections,
            "total_latency_ms": round(total_latency, 2),
            "avg_latency_ms": round(total_latency / max(len(results["events"]), 1), 2),
            "expected_hit": list(expected & found),
            "expected_miss": list(expected - found),
        }

    return results


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------

def print_summary_table(all_results: list) -> None:
    versions = ["v1", "v2", "v3", "v4", "v5"]
    col = 16

    header = f"{'Scenario':<28}" + "".join(f"  {v:<{col}}" for v in versions)
    print("\n" + "=" * len(header))
    print(header)
    print("=" * len(header))

    for r in all_results:
        name = r["scenario"][:27]
        expected = r["expected_detections"]

        row = f"{name:<28}"
        for v in versions:
            if v not in r["summary"]:
                row += f"  {'N/A':<{col}}"
                continue
            s = r["summary"][v]
            hit = len(s["expected_hit"])
            miss = len(s["expected_miss"])
            total_exp = hit + miss
            fired_str = ",".join(s["detectors_fired"][:2]) or "-"
            status = f"{hit}/{total_exp}hit {fired_str}"
            row += f"  {status:<{col}}"
        print(row)

    print("=" * len(header))


def save_results(all_results: list, scenario_filter: str = None) -> str:
    ts = time.strftime("%Y%m%d-%H%M%S")
    suffix = f"-{scenario_filter}" if scenario_filter else "-all"
    out_path = RESULTS_DIR / f"eval{suffix}-{ts}.json"
    with open(out_path, "w") as f:
        json.dump(all_results, f, indent=2)
    return str(out_path)


# ---------------------------------------------------------------------------
# Version loader with retry-safe import isolation
# ---------------------------------------------------------------------------

def load_all_versions() -> dict:
    versions = {}
    for vname in ("v1", "v2", "v3", "v4", "v5"):
        try:
            versions[vname] = load_version(vname)
        except Exception as e:
            print(f"[WARN] Could not load {vname}: {e}", file=sys.stderr)
    return versions


# ---------------------------------------------------------------------------
# Coverage map
# ---------------------------------------------------------------------------

def coverage_map(all_results: list) -> dict:
    """Build a map of which detectors are covered (fired at least once) across scenarios.
    Returns:
        {
            "detector_name": {
                "covered": bool,
                "scenarios": ["scenario_name", ...],  # where it fired
                "versions": ["v1", ...]               # which versions triggered it
            }
        }
    """
    all_detectors = [
        "LOOP", "READ-STORM", "ERROR-CASCADE", "DRIFT", "YOLO",
        "THRASH", "STALL", "ANTI-PATTERN", "SKILL-SUGGEST",
        "ENFORCEMENT", "SIDECAR GATE", "GATE-BLOCK",
        "autonomous_git", "force_push", "rm_rf", "secret_exposure",
    ]

    coverage: dict = {det: {"covered": False, "scenarios": [], "versions": []} for det in all_detectors}

    for r in all_results:
        for vname, summary in r.get("summary", {}).items():
            for det in summary.get("detectors_fired", []):
                if det not in coverage:
                    coverage[det] = {"covered": False, "scenarios": [], "versions": []}
                entry = coverage[det]
                entry["covered"] = True
                if r["scenario"] not in entry["scenarios"]:
                    entry["scenarios"].append(r["scenario"])
                if vname not in entry["versions"]:
                    entry["versions"].append(vname)

    return coverage


def print_coverage_map(cmap: dict) -> None:
    covered = [d for d, v in cmap.items() if v["covered"]]
    uncovered = [d for d, v in cmap.items() if not v["covered"]]

    print(f"\nDetector Coverage: {len(covered)}/{len(cmap)} covered")
    print("-" * 60)
    for det in sorted(covered):
        entry = cmap[det]
        print(f"  [OK] {det:<22}  versions={entry['versions']}  scenarios={len(entry['scenarios'])}")
    for det in sorted(uncovered):
        print(f"  [--] {det:<22}  NOT COVERED")
    print()


# ---------------------------------------------------------------------------
# Regression comparison
# ---------------------------------------------------------------------------

def compare_to_baseline(current: list, baseline_path: str) -> None:
    """Compare current results against a saved baseline, print regressions."""
    try:
        with open(baseline_path) as f:
            baseline_results = json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        print(f"[ERROR] Could not load baseline {baseline_path!r}: {e}", file=sys.stderr)
        return

    # Index baseline by scenario name
    baseline_index = {r["scenario"]: r for r in baseline_results}

    regressions = []
    improvements = []
    new_scenarios = []

    for r in current:
        name = r["scenario"]
        if name not in baseline_index:
            new_scenarios.append(name)
            continue

        b = baseline_index[name]
        for vname in r["summary"]:
            if vname not in b.get("summary", {}):
                continue
            cur_s = r["summary"][vname]
            base_s = b["summary"][vname]

            cur_hit = set(cur_s["expected_hit"])
            base_hit = set(base_s["expected_hit"])
            lost = base_hit - cur_hit
            gained = cur_hit - base_hit

            if lost:
                regressions.append({
                    "scenario": name,
                    "version": vname,
                    "lost_detections": sorted(lost),
                })
            if gained:
                improvements.append({
                    "scenario": name,
                    "version": vname,
                    "gained_detections": sorted(gained),
                })

    print(f"\nRegression Report (baseline: {baseline_path})")
    print("=" * 60)
    if regressions:
        print(f"REGRESSIONS ({len(regressions)}):")
        for reg in regressions:
            print(f"  [{reg['version']}] {reg['scenario']}: lost {reg['lost_detections']}")
    else:
        print("No regressions.")

    if improvements:
        print(f"\nIMPROVEMENTS ({len(improvements)}):")
        for imp in improvements:
            print(f"  [{imp['version']}] {imp['scenario']}: gained {imp['gained_detections']}")

    if new_scenarios:
        print(f"\nNew scenarios (not in baseline): {new_scenarios}")
    print()


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    args = sys.argv[1:]

    if "--list" in args:
        scenario_files = sorted(SCENARIOS_DIR.glob("*.json"))
        if not scenario_files:
            print("No scenarios found in", SCENARIOS_DIR)
        else:
            print(f"Available scenarios ({len(scenario_files)}):")
            for sf in scenario_files:
                with open(sf) as f:
                    sc = json.load(f)
                exp = sc.get("expected_detections", [])
                print(f"  {sf.stem:<30}  expect: {exp}")
        return

    # Parse --baseline / --compare / --coverage flags
    baseline_save_path = None
    compare_path = None
    show_coverage = "--coverage" in args

    if "--baseline" in args:
        idx = args.index("--baseline")
        if idx + 1 < len(args):
            baseline_save_path = args[idx + 1]

    if "--compare" in args:
        idx = args.index("--compare")
        if idx + 1 < len(args):
            compare_path = args[idx + 1]

    # Filter to single scenario if given (first non-flag arg)
    scenario_filter = None
    for arg in args:
        if not arg.startswith("--"):
            scenario_filter = arg
            break

    scenario_files = sorted(SCENARIOS_DIR.glob("*.json"))
    if not scenario_files:
        print("No scenario files found in", SCENARIOS_DIR)
        sys.exit(1)

    if scenario_filter:
        scenario_files = [sf for sf in scenario_files if sf.stem == scenario_filter]
        if not scenario_files:
            print(f"Scenario '{scenario_filter}' not found. Use --list to see available.")
            sys.exit(1)

    print(f"Loading versions...")
    versions = load_all_versions()
    if not versions:
        print("No versions could be loaded.")
        sys.exit(1)
    print(f"Loaded: {list(versions.keys())}")

    all_results = []
    for sf in scenario_files:
        print(f"\nRunning: {sf.stem} ({len(json.load(open(sf))['events'])} events)...", end=" ")
        try:
            result = run_scenario(sf, versions)
            all_results.append(result)
            # Quick inline summary
            summaries = []
            for v, s in result["summary"].items():
                hit = len(s["expected_hit"])
                total = hit + len(s["expected_miss"])
                summaries.append(f"{v}:{hit}/{total}")
            print(" | ".join(summaries))
        except Exception as e:
            print(f"ERROR: {e}")
            import traceback
            traceback.print_exc()

    if all_results:
        print_summary_table(all_results)
        out_path = save_results(all_results, scenario_filter)
        print(f"\nResults saved to: {out_path}")

        # Coverage map
        if show_coverage:
            cmap = coverage_map(all_results)
            print_coverage_map(cmap)

        # Save as baseline
        if baseline_save_path:
            import shutil
            shutil.copy(out_path, baseline_save_path)
            print(f"Baseline saved to: {baseline_save_path}")

        # Compare against baseline
        if compare_path:
            compare_to_baseline(all_results, compare_path)


if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""Sidecar Eval Runner — replays scenario events through v1/v2/v3/v4/v5.

	Usage:
	python eval_runner.py # run all scenarios
	python eval_runner.py loop_detection # run single scenario by name
	python eval_runner.py --list # list available scenarios
	python eval_runner.py --baseline results/X.json # save current run as baseline
	python eval_runner.py --compare results/X.json # compare against saved baseline
	python eval_runner.py --coverage # show detector coverage map
	"""

	import copy
	import glob
	import importlib.util
	import json
	import os
	import sys
	import time
	from pathlib import Path
	from unittest.mock import MagicMock, patch

	VERSIONS_DIR = Path(__file__).parent / "versions"
	SCENARIOS_DIR = Path(__file__).parent / "scenarios"
	RESULTS_DIR = Path(__file__).parent / "results"

	RESULTS_DIR.mkdir(exist_ok=True)

	# In-memory session state store — keyed by (version_name, session_id)
	_SESSION_STORE: dict = {}


	# ---------------------------------------------------------------------------
	# Version loading
	# ---------------------------------------------------------------------------

	def load_version(version_name: str):
	"""Import sidecar module from versions/{version_name}/ with all dependencies mocked."""
	version_dir = VERSIONS_DIR / version_name
	sidecar_path = version_dir / "sidecar.py"

	if not sidecar_path.exists():
	raise FileNotFoundError(f"No sidecar.py in {version_dir}")

	# Add version dir to front of sys.path so relative imports work
	version_str = str(version_dir)
	if version_str not in sys.path:
	sys.path.insert(0, version_str)

	# Module name must be unique per version to avoid caching collisions
	mod_name = f"sidecar_{version_name}"

	# Remove cached version if present (allows re-import with fresh state)
	for key in list(sys.modules.keys()):
	if key.startswith(f"sidecar_{version_name}") or (
	version_str in sys.modules.get(key, object).__spec__.origin
	if hasattr(sys.modules.get(key, None), "__spec__") and
	sys.modules[key].__spec__ is not None and
	sys.modules[key].__spec__.origin
	else False
	):
	del sys.modules[key]

	spec = importlib.util.spec_from_file_location(mod_name, sidecar_path)
	module = importlib.util.module_from_spec(spec)

	# Patch companion file reads to return empty strings (no actual files needed)
	mock_open = _make_mock_open()

	patches = [
	# Companion files — mock open() in the sidecar module's builtins
	patch("builtins.open", mock_open),
	# fcntl.flock — no-op (no file locking in eval)
	patch("fcntl.flock", return_value=None),
	# tempfile.mkstemp — return a fake fd/path pair (save goes through mock open)
	patch("tempfile.mkstemp", side_effect=_fake_mkstemp),
	# os.replace — no-op (we use in-memory state)
	patch("os.replace", return_value=None),
	# os.unlink — no-op
	patch("os.unlink", return_value=None),
	# os.fdopen — return mock file object
	patch("os.fdopen", side_effect=_fake_fdopen),
	]

	# For v2/v3/v4/v5: mock SessionWatcher so it doesn't start threads
	if version_name in ("v2", "v3", "v4", "v5"):
	patches.append(patch.dict(sys.modules, {
	"watchdog": MagicMock(),
	"watchdog.observers": MagicMock(),
	"watchdog.events": MagicMock(),
	}))

	for p in patches:
	p.start()

	try:
	spec.loader.exec_module(module)
	finally:
	for p in patches:
	try:
	p.stop()
	except RuntimeError:
	pass # already stopped

	# Tag module so we can reference its version
	module._eval_version = version_name
	module._eval_version_dir = version_dir

	return module


	def _make_mock_open():
	"""Return a mock open() that returns empty content for companion files,
	and delegates lock files / tmp files to a no-op in-memory implementation."""
	real_open = open

	class _FakeFile:
	def __init__(self):
	self._data = ""
	def read(self):
	return self._data
	def write(self, data):
	pass
	def __enter__(self):
	return self
	def __exit__(self, *a):
	pass
	def fileno(self):
	return -1

	def _mock_open(path, mode="r", args, *kwargs):
	path_str = str(path) if not isinstance(path, str) else path
	# Companion files and coaching rules — return empty
	if any(name in path_str for name in [
	"WelcheFehlerVermeiden", "Skilluebersicht", "coaching_rules",
	"minimax-key", "gemini-key", "sidecar-rules.json",
	]):
	return _FakeFile()
	# Lock files and state files in /tmp — return no-op file
	if path_str.startswith("/tmp/sidecar"):
	return _FakeFile()
	# Log file — no-op
	if "claude-sidecar.log" in path_str:
	return _FakeFile()
	# For actual eval/version source files, use real open
	return real_open(path, mode, args, *kwargs)

	return _mock_open


	_fake_fd_counter = [100]

	def _fake_mkstemp(**kwargs):
	fd = _fake_fd_counter[0]
	_fake_fd_counter[0] += 1
	return (fd, f"/tmp/fake-sidecar-{fd}.tmp")


	def _fake_fdopen(fd, mode="r", args, *kwargs):
	class _NoopFile:
	def write(self, d): pass
	def __enter__(self): return self
	def __exit__(self, *a): pass
	return _NoopFile()


	# ---------------------------------------------------------------------------
	# State creation with in-memory session I/O patches
	# ---------------------------------------------------------------------------

	def create_fresh_state(version_name: str, module):
	"""Create a fresh SidecarState with file I/O replaced by in-memory dicts."""

	# Per-version session store — isolated from other versions
	store: dict = {}

	def _load_session(session_id: str) -> dict:
	if session_id not in store:
	store[session_id] = {
	"history": [],
	"last_analysis_ts": 0,
	"session_start_ts": 0,
	"files_touched": [],
	"total_calls": 0,
	"recap_done": False,
	}
	return copy.deepcopy(store[session_id])

	def _save_session(session_id: str, state: dict) -> None:
	store[session_id] = copy.deepcopy(state)

	def _lock_path_noop(session_id: str) -> str:
	return f"/tmp/eval-lock-{session_id}.lock"

	# Patch module-level functions
	module._load_session_state = _load_session
	module._save_session_state = _save_session
	module._lock_path = _lock_path_noop

	# Build state object — companion file loads fail gracefully (mock open returns "")
	mock_open = _make_mock_open()
	with patch("builtins.open", mock_open), \
	patch("fcntl.flock", return_value=None):
	state = module.SidecarState()
	# Load companion files (returns empty — no actual files needed)
	state.load_companion_files()

	# v3: disable LLM judge — no external calls
	if hasattr(state, "llm_judge"):
	state.llm_judge.enabled = False

	# v2/v3: disable SessionWatcher — stop background threads if started
	if hasattr(state, "session_watcher"):
	try:
	state.session_watcher.stop()
	except Exception:
	pass
	# Replace with a no-op stub
	state.session_watcher = MagicMock()

	return state, store


	# ---------------------------------------------------------------------------
	# Event processing
	# ---------------------------------------------------------------------------

	def process_event_for_version(module, event: dict, state) -> dict:
	"""Run _process_event on a version, return response + latency."""
	full_event = {
	"session_id": event.get("session_id", "eval-session-001"),
	"tool_name": event["tool"],
	"tool_input": event.get("input", {}),
	"error": "Error occurred" if event.get("had_error") else "",
	"hook_event_name": event.get("hook_event", "PostToolUse"),
	"tool_response": event.get("response_text", ""),
	}

	# Disable cooldown in eval so all detections are captured regardless of timing
	version_mod_name = f"sidecar_{getattr(module, '_eval_version', 'vX')}"
	with patch("fcntl.flock", return_value=None), \
	patch("builtins.open", _make_mock_open()), \
	patch("tempfile.mkstemp", side_effect=_fake_mkstemp), \
	patch("os.replace", return_value=None), \
	patch("os.fdopen", side_effect=_fake_fdopen), \
	patch.object(module, "CONCERN_COOLDOWN_SECONDS", 0):
	start = time.perf_counter()
	result = module._process_event(full_event, state)
	elapsed = time.perf_counter() - start

	return {
	"response": result,
	"latency_ms": round(elapsed * 1000, 2),
	"fired": _extract_fired_detectors(result),
	}


	def _extract_fired_detectors(response: dict) -> list:
	"""Parse which detectors fired from the response dict."""
	fired = []
	if not response:
	return fired

	text = response.get("additionalContext", "") + response.get("systemMessage", "")
	reason = response.get("reason", "")
	all_text = text + " " + reason

	detector_names = [
	"LOOP", "READ-STORM", "ERROR-CASCADE", "DRIFT", "YOLO",
	"THRASH", "STALL", "ANTI-PATTERN", "SKILL-SUGGEST",
	"ENFORCEMENT", "SIDECAR GATE",
	]
	for det in detector_names:
	if det in all_text:
	fired.append(det)

	if response.get("decision") == "block":
	fired.append("GATE-BLOCK")
	# Extract gate rule name from reason: "[SIDECAR GATE] gate_name: message"
	if "GATE]" in reason:
	gate_part = reason.split("GATE]")[-1].strip()
	gate_name = gate_part.split(":")[0].strip()
	# Map gate names to expected detection names
	gate_map = {
	"gate_autonomous_git": "autonomous_git",
	"gate_force_push": "force_push",
	"gate_rm_rf": "rm_rf",
	"gate_secret_exposure": "secret_exposure",
	}
	mapped = gate_map.get(gate_name, gate_name)
	if mapped not in fired:
	fired.append(mapped)

	# Extract enforcement rule names: "ENFORCEMENT [rule_name]:"
	import re
	for m in re.finditer(r"ENFORCEMENT \[(\w+)\]", all_text):
	rule_name = m.group(1)
	if rule_name not in fired:
	fired.append(rule_name)

	return fired


	# ---------------------------------------------------------------------------
	# Scenario runner
	# ---------------------------------------------------------------------------

	def run_scenario(scenario_path: Path, versions: dict) -> dict:
	"""Run one scenario through all versions, return results dict."""
	with open(scenario_path) as f:
	scenario = json.load(f)

	results = {
	"scenario": scenario["name"],
	"description": scenario.get("description", ""),
	"expected_detections": scenario.get("expected_detections", []),
	"events": [],
	"summary": {},
	}

	# Create fresh state per version (isolated stores)
	states = {}
	for vname, vmod in versions.items():
	state_obj, _ = create_fresh_state(vname, vmod)
	states[vname] = (vmod, state_obj)

	for i, event in enumerate(scenario["events"]):
	event_result = {
	"index": i,
	"tool": event["tool"],
	"input": event.get("input", {}),
	"had_error": event.get("had_error", False),
	"responses": {},
	}
	for vname, (vmod, state) in states.items():
	event_result["responses"][vname] = process_event_for_version(vmod, event, state)

	results["events"].append(event_result)

	# Build per-version summary
	for vname in versions:
	all_fired = []
	total_latency = 0.0
	injections = 0
	for ev in results["events"]:
	r = ev["responses"][vname]
	all_fired.extend(r["fired"])
	total_latency += r["latency_ms"]
	if r["response"]:
	injections += 1

	expected = set(results["expected_detections"])
	found = set(all_fired)
	results["summary"][vname] = {
	"detectors_fired": list(dict.fromkeys(all_fired)), # deduplicated, ordered
	"injections": injections,
	"total_latency_ms": round(total_latency, 2),
	"avg_latency_ms": round(total_latency / max(len(results["events"]), 1), 2),
	"expected_hit": list(expected & found),
	"expected_miss": list(expected - found),
	}

	return results


	# ---------------------------------------------------------------------------
	# Output formatting
	# ---------------------------------------------------------------------------

	def print_summary_table(all_results: list) -> None:
	versions = ["v1", "v2", "v3", "v4", "v5"]
	col = 16

	header = f"{'Scenario':<28}" + "".join(f" {v:<{col}}" for v in versions)
	print("\n" + "=" * len(header))
	print(header)
	print("=" * len(header))

	for r in all_results:
	name = r["scenario"][:27]
	expected = r["expected_detections"]

	row = f"{name:<28}"
	for v in versions:
	if v not in r["summary"]:
	row += f" {'N/A':<{col}}"
	continue
	s = r["summary"][v]
	hit = len(s["expected_hit"])
	miss = len(s["expected_miss"])
	total_exp = hit + miss
	fired_str = ",".join(s["detectors_fired"][:2]) or "-"
	status = f"{hit}/{total_exp}hit {fired_str}"
	row += f" {status:<{col}}"
	print(row)

	print("=" * len(header))


	def save_results(all_results: list, scenario_filter: str = None) -> str:
	ts = time.strftime("%Y%m%d-%H%M%S")
	suffix = f"-{scenario_filter}" if scenario_filter else "-all"
	out_path = RESULTS_DIR / f"eval{suffix}-{ts}.json"
	with open(out_path, "w") as f:
	json.dump(all_results, f, indent=2)
	return str(out_path)


	# ---------------------------------------------------------------------------
	# Version loader with retry-safe import isolation
	# ---------------------------------------------------------------------------

	def load_all_versions() -> dict:
	versions = {}
	for vname in ("v1", "v2", "v3", "v4", "v5"):
	try:
	versions[vname] = load_version(vname)
	except Exception as e:
	print(f"[WARN] Could not load {vname}: {e}", file=sys.stderr)
	return versions


	# ---------------------------------------------------------------------------
	# Coverage map
	# ---------------------------------------------------------------------------

	def coverage_map(all_results: list) -> dict:
	"""Build a map of which detectors are covered (fired at least once) across scenarios.

	Returns:
	{
	"detector_name": {
	"covered": bool,
	"scenarios": ["scenario_name", ...], # where it fired
	"versions": ["v1", ...] # which versions triggered it
	}
	}
	"""
	all_detectors = [
	"LOOP", "READ-STORM", "ERROR-CASCADE", "DRIFT", "YOLO",
	"THRASH", "STALL", "ANTI-PATTERN", "SKILL-SUGGEST",
	"ENFORCEMENT", "SIDECAR GATE", "GATE-BLOCK",
	"autonomous_git", "force_push", "rm_rf", "secret_exposure",
	]

	coverage: dict = {det: {"covered": False, "scenarios": [], "versions": []} for det in all_detectors}

	for r in all_results:
	for vname, summary in r.get("summary", {}).items():
	for det in summary.get("detectors_fired", []):
	if det not in coverage:
	coverage[det] = {"covered": False, "scenarios": [], "versions": []}
	entry = coverage[det]
	entry["covered"] = True
	if r["scenario"] not in entry["scenarios"]:
	entry["scenarios"].append(r["scenario"])
	if vname not in entry["versions"]:
	entry["versions"].append(vname)

	return coverage


	def print_coverage_map(cmap: dict) -> None:
	covered = [d for d, v in cmap.items() if v["covered"]]
	uncovered = [d for d, v in cmap.items() if not v["covered"]]

	print(f"\nDetector Coverage: {len(covered)}/{len(cmap)} covered")
	print("-" * 60)
	for det in sorted(covered):
	entry = cmap[det]
	print(f" [OK] {det:<22} versions={entry['versions']} scenarios={len(entry['scenarios'])}")
	for det in sorted(uncovered):
	print(f" [--] {det:<22} NOT COVERED")
	print()


	# ---------------------------------------------------------------------------
	# Regression comparison
	# ---------------------------------------------------------------------------

	def compare_to_baseline(current: list, baseline_path: str) -> None:
	"""Compare current results against a saved baseline, print regressions."""
	try:
	with open(baseline_path) as f:
	baseline_results = json.load(f)
	except (OSError, json.JSONDecodeError) as e:
	print(f"[ERROR] Could not load baseline {baseline_path!r}: {e}", file=sys.stderr)
	return

	# Index baseline by scenario name
	baseline_index = {r["scenario"]: r for r in baseline_results}

	regressions = []
	improvements = []
	new_scenarios = []

	for r in current:
	name = r["scenario"]
	if name not in baseline_index:
	new_scenarios.append(name)
	continue

	b = baseline_index[name]
	for vname in r["summary"]:
	if vname not in b.get("summary", {}):
	continue
	cur_s = r["summary"][vname]
	base_s = b["summary"][vname]

	cur_hit = set(cur_s["expected_hit"])
	base_hit = set(base_s["expected_hit"])
	lost = base_hit - cur_hit
	gained = cur_hit - base_hit

	if lost:
	regressions.append({
	"scenario": name,
	"version": vname,
	"lost_detections": sorted(lost),
	})
	if gained:
	improvements.append({
	"scenario": name,
	"version": vname,
	"gained_detections": sorted(gained),
	})

	print(f"\nRegression Report (baseline: {baseline_path})")
	print("=" * 60)
	if regressions:
	print(f"REGRESSIONS ({len(regressions)}):")
	for reg in regressions:
	print(f" [{reg['version']}] {reg['scenario']}: lost {reg['lost_detections']}")
	else:
	print("No regressions.")

	if improvements:
	print(f"\nIMPROVEMENTS ({len(improvements)}):")
	for imp in improvements:
	print(f" [{imp['version']}] {imp['scenario']}: gained {imp['gained_detections']}")

	if new_scenarios:
	print(f"\nNew scenarios (not in baseline): {new_scenarios}")
	print()


	# ---------------------------------------------------------------------------
	# Main
	# ---------------------------------------------------------------------------

	def main():
	args = sys.argv[1:]

	if "--list" in args:
	scenario_files = sorted(SCENARIOS_DIR.glob("*.json"))
	if not scenario_files:
	print("No scenarios found in", SCENARIOS_DIR)
	else:
	print(f"Available scenarios ({len(scenario_files)}):")
	for sf in scenario_files:
	with open(sf) as f:
	sc = json.load(f)
	exp = sc.get("expected_detections", [])
	print(f" {sf.stem:<30} expect: {exp}")
	return

	# Parse --baseline / --compare / --coverage flags
	baseline_save_path = None
	compare_path = None
	show_coverage = "--coverage" in args

	if "--baseline" in args:
	idx = args.index("--baseline")
	if idx + 1 < len(args):
	baseline_save_path = args[idx + 1]

	if "--compare" in args:
	idx = args.index("--compare")
	if idx + 1 < len(args):
	compare_path = args[idx + 1]

	# Filter to single scenario if given (first non-flag arg)
	scenario_filter = None
	for arg in args:
	if not arg.startswith("--"):
	scenario_filter = arg
	break

	scenario_files = sorted(SCENARIOS_DIR.glob("*.json"))
	if not scenario_files:
	print("No scenario files found in", SCENARIOS_DIR)
	sys.exit(1)

	if scenario_filter:
	scenario_files = [sf for sf in scenario_files if sf.stem == scenario_filter]
	if not scenario_files:
	print(f"Scenario '{scenario_filter}' not found. Use --list to see available.")
	sys.exit(1)

	print(f"Loading versions...")
	versions = load_all_versions()
	if not versions:
	print("No versions could be loaded.")
	sys.exit(1)
	print(f"Loaded: {list(versions.keys())}")

	all_results = []
	for sf in scenario_files:
	print(f"\nRunning: {sf.stem} ({len(json.load(open(sf))['events'])} events)...", end=" ")
	try:
	result = run_scenario(sf, versions)
	all_results.append(result)
	# Quick inline summary
	summaries = []
	for v, s in result["summary"].items():
	hit = len(s["expected_hit"])
	total = hit + len(s["expected_miss"])
	summaries.append(f"{v}:{hit}/{total}")
	print(" \| ".join(summaries))
	except Exception as e:
	print(f"ERROR: {e}")
	import traceback
	traceback.print_exc()

	if all_results:
	print_summary_table(all_results)
	out_path = save_results(all_results, scenario_filter)
	print(f"\nResults saved to: {out_path}")

	# Coverage map
	if show_coverage:
	cmap = coverage_map(all_results)
	print_coverage_map(cmap)

	# Save as baseline
	if baseline_save_path:
	import shutil
	shutil.copy(out_path, baseline_save_path)
	print(f"Baseline saved to: {baseline_save_path}")

	# Compare against baseline
	if compare_path:
	compare_to_baseline(all_results, compare_path)


	if __name__ == "__main__":
	main()