turn-1

.claude/hooks/capture_session_event.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""
+Hook to capture session start/end events.
+Cross-platform support for Windows, macOS, and Linux.
+"""
+import json
+import sys
+import os
+import subprocess
+from datetime import datetime, timezone
+from claude_code_capture_utils import get_log_file_path, add_ab_metadata
+
+def get_git_metadata(repo_dir):
+    """Get current git commit and branch."""
+    try:
+        # Get current commit hash
+        commit_result = subprocess.run(
+            ['git', 'rev-parse', 'HEAD'],
+            cwd=repo_dir,
+            capture_output=True,
+            text=True,
+            timeout=10
+        )
+
+        # Get current branch
+        branch_result = subprocess.run(
+            ['git', 'branch', '--show-current'],
+            cwd=repo_dir,
+            capture_output=True,
+            text=True,
+            timeout=10
+        )
+
+        git_metadata = {
+            "base_commit": commit_result.stdout.strip() if commit_result.returncode == 0 else None,
+            "branch": branch_result.stdout.strip() if branch_result.returncode == 0 else None,
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        if git_metadata["base_commit"]:
+            return git_metadata
+        else:
+            return None
+
+    except Exception as e:
+        print(f"Warning: Could not capture git metadata: {e}", file=sys.stderr)
+        return None
+
+def main():
+    try:
+        if len(sys.argv) < 2:
+            print("Usage: capture_session_event.py [start|end]", file=sys.stderr)
+            sys.exit(1)
+
+        event_type = sys.argv[1].lower()
+        if event_type not in ["start", "end"]:
+            print("Event type must be 'start' or 'end'", file=sys.stderr)
+            sys.exit(1)
+
+        input_data = json.load(sys.stdin)
+
+        session_id = input_data.get("session_id", "unknown")
+        transcript_path = input_data.get("transcript_path", "")
+        cwd = input_data.get("cwd", "")
+
+        if event_type == "start":
+            # Session start: capture git metadata
+            git_metadata = get_git_metadata(cwd)
+
+            log_entry = {
+                "type": "session_start",
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "session_id": session_id,
+                "transcript_path": transcript_path,
+                "cwd": cwd,
+                "git_metadata": git_metadata
+            }
+
+            log_entry = add_ab_metadata(log_entry, cwd)
+
+            if git_metadata:
+                print(f"[OK] Captured git metadata: {git_metadata['base_commit'][:8]} on {git_metadata['branch']}")
+
+            # Write session_start event
+            log_file = get_log_file_path(session_id, cwd)
+            with open(log_file, "a", encoding="utf-8") as f:
+                f.write(json.dumps(log_entry) + "\n")
+
+        elif event_type == "end":
+            # Session end: log the event
+            log_entry = {
+                "type": "session_end",
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "session_id": session_id,
+                "transcript_path": transcript_path,
+                "cwd": cwd,
+                "reason": input_data.get("reason", "")
+            }
+
+            log_entry = add_ab_metadata(log_entry, cwd)
+
+            # Write session_end event
+            log_file = get_log_file_path(session_id, cwd)
+            with open(log_file, "a", encoding="utf-8") as f:
+                f.write(json.dumps(log_entry) + "\n")
+
+    except Exception as e:
+        print(f"[ERROR] Session {event_type}: {e}", file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()

.claude/hooks/claude_code_capture_utils.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""
+Utility functions for A/B testing hooks.
+Cross-platform support for Windows, macOS, and Linux.
+"""
+import json
+import os
+from pathlib import Path
+
+def detect_model_lane(cwd):
+    """Detect if we're in model_a or model_b directory."""
+    path_parts = Path(cwd).parts
+    if 'model_a' in path_parts:
+        return 'model_a'
+    elif 'model_b' in path_parts:
+        return 'model_b'
+    return None
+
+def get_experiment_root(cwd):
+    """Get the experiment root directory (parent of model_a/model_b)."""
+    current_path = Path(cwd)
+
+    # Check if we're inside a model_a or model_b directory
+    # Look for the parent that contains both model_a and model_b
+    for parent in [current_path] + list(current_path.parents):
+        if (parent / 'model_a').exists() and (parent / 'model_b').exists():
+            return str(parent)
+        # Also check one level up (in case we're inside the cloned repo)
+        parent_up = parent.parent
+        if (parent_up / 'model_a').exists() and (parent_up / 'model_b').exists():
+            return str(parent_up)
+    return None
+
+def read_manifest(experiment_root):
+    """Read the manifest.json file to get task_id and model assignments."""
+    try:
+        manifest_path = os.path.join(experiment_root, 'manifest.json')
+        if os.path.exists(manifest_path):
+            with open(manifest_path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+    except Exception:
+        pass
+    return {}
+
+def get_ab_metadata(cwd):
+    """Get A/B testing metadata (task_id, model_lane, model_name) from current directory."""
+    model_lane = detect_model_lane(cwd)
+    experiment_root = get_experiment_root(cwd)
+
+    if not model_lane or not experiment_root:
+        return {}
+
+    manifest = read_manifest(experiment_root)
+
+    metadata = {
+        "task_id": manifest.get("task_id"),
+        "model_lane": model_lane,
+        "experiment_root": experiment_root
+    }
+
+    # Get model name from assignments
+    assignments = manifest.get("assignments", {})
+    if model_lane in assignments:
+        metadata["model_name"] = assignments[model_lane]
+
+    return metadata
+
+def get_log_file_path(session_id, cwd):
+    """Get the correct log file path for A/B testing (routes to model-specific directory)."""
+    model_lane = detect_model_lane(cwd)
+    experiment_root = get_experiment_root(cwd)
+
+    if model_lane and experiment_root:
+        # Route to model-specific logs directory
+        logs_dir = os.path.join(experiment_root, "logs", model_lane)
+        os.makedirs(logs_dir, exist_ok=True)
+        return os.path.join(logs_dir, f"session_{session_id}.jsonl")
+    else:
+        # Fallback to current behavior
+        project_dir = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
+        logs_dir = os.path.join(project_dir, "logs")
+        os.makedirs(logs_dir, exist_ok=True)
+        return os.path.join(logs_dir, f"session_{session_id}.jsonl")
+
+def add_ab_metadata(event, cwd):
+    """Add A/B testing metadata to an event."""
+    ab_metadata = get_ab_metadata(cwd)
+    if ab_metadata:
+        for key, value in ab_metadata.items():
+            if value is not None:
+                event[key] = value
+    return event
+