Session Feedback Analyzer

#!/usr/bin/env python3
"""Session feedback analyzer for the auto-improvement pipeline.

Parses Claude Code session JSONL files, detects skill invocations,
classifies user responses (correction/acceptance/partial) within a
3-turn influence window, and outputs feedback.jsonl for the generator.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
from dataclasses import dataclass, asdict
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any, Iterator

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from lib.common import utc_now_iso, write_json  # noqa: E402

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

CORRECTION_KEYWORDS_ZH = ("不对", "错了", "不是这样", "重新来", "换个方案", "不行", "有问题")
CORRECTION_KEYWORDS_EN = ("wrong", "incorrect", "no,", "no ", "redo", "try again", "that's not")
REDO_KEYWORDS = ("重新来", "redo", "try again", "换个方案", "再来一次", "重做")
ACCEPTANCE_KEYWORDS_ZH = ("好", "可以", "对的", "继续", "没问题", "行", "好的")
ACCEPTANCE_KEYWORDS_EN = ("looks good", "lgtm", "perfect", "correct", "yes", "great", "thanks")
REVERT_COMMANDS = ("git checkout", "git restore", "git reset")

DIMENSION_KEYWORDS = {
    "accuracy": ("naming", "format", "style", "命名", "格式", "风格", "拼写", "typo"),
    "coverage": ("missing", "forgot", "没考虑", "缺少", "漏了", "incomplete"),
    "reliability": ("again", "又", "重复", "inconsistent", "不稳定"),
    "efficiency": ("slow", "verbose", "太慢", "太多", "太长", "冗余"),
    "security": ("security", "secret", "安全", "密钥", "token", "credential"),
    "trigger_quality": ("wrong skill", "不该触发", "shouldn't trigger", "错误的skill"),
}

INFLUENCE_WINDOW = 3  # max user turns to scan after skill invocation


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------


@dataclass
class SkillInvocation:
    invocation_id: str
    skill_id: str
    timestamp: str
    message_index: int


@dataclass
class FeedbackEvent:
    event_id: str
    timestamp: str
    session_id: str
    skill_id: str
    invocation_uuid: str
    outcome: str  # "correction" | "acceptance" | "partial"
    confidence: float
    correction_type: str | None  # "rejection" | "revert" | "redo" | "partial" | None
    user_message_snippet: str
    turns_to_feedback: int
    ai_tools_used: list[str]
    dimension_hint: str | None


# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Analyze Claude Code sessions for skill feedback signals",
    )
    parser.add_argument(
        "--session-dir",
        default=str(Path.home() / ".claude" / "projects"),
        help="Root directory for session JSONL files",
    )
    parser.add_argument(
        "--output",
        default="feedback-store/feedback.jsonl",
        help="Output path for feedback JSONL",
    )
    parser.add_argument(
        "--no-snippets",
        action="store_true",
        help="Omit user message snippets from output",
    )
    parser.add_argument(
        "--skill-filter",
        help="Only analyze invocations of this skill",
    )
    parser.add_argument(
        "--min-invocations",
        type=int,
        default=5,
        help="Minimum invocations before computing metrics",
    )
    return parser.parse_args(argv)


def iter_session_files(session_dir: Path) -> Iterator[Path]:
    """Yield session JSONL files, skipping test/tmp directories."""
    for path in session_dir.rglob("*.jsonl"):
        path_str = str(path)
        if "pytest" in path_str or "/tmp/" in path_str:
            continue
        if "/subagents/" in path_str:
            continue
        yield path


def parse_session(path: Path) -> list[dict[str, Any]]:
    """Parse a session JSONL file into a list of message dicts."""
    messages: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                messages.append(json.loads(line))
            except json.JSONDecodeError:
                continue  # skip malformed lines
    return messages


def extract_session_id(path: Path) -> str:
    """Extract session ID from the JSONL filename."""
    return path.stem


def extract_project(path: Path) -> str:
    """Extract project name from the directory structure."""
    return path.parent.name


# ---------------------------------------------------------------------------
# Skill invocation detection
# ---------------------------------------------------------------------------


def detect_skill_invocations(messages: list[dict[str, Any]]) -> list[SkillInvocation]:
    """Detect skill invocations via tool_use or slash commands."""
    invocations: list[SkillInvocation] = []

    for idx, entry in enumerate(messages):
        msg_type = entry.get("type")
        uuid = entry.get("uuid", "")
        timestamp = entry.get("timestamp", "")

        # Path A: assistant tool_use with name=="Skill"
        if msg_type == "assistant":
            content = entry.get("message", {}).get("content", [])
            if isinstance(content, list):
                for block in content:
                    if (isinstance(block, dict)
                            and block.get("type") == "tool_use"
                            and block.get("name") == "Skill"):
                        skill_id = block.get("input", {}).get("skill", "")
                        if skill_id:
                            invocations.append(SkillInvocation(
                                invocation_id=uuid,
                                skill_id=skill_id,
                                timestamp=timestamp,
                                message_index=idx,
                            ))

        # Path B: system local_command with <command-name>
        if msg_type == "system" and entry.get("subtype") == "local_command":
            content_str = str(entry.get("content", ""))
            match = re.search(r"<command-name>/?([\w-]+)</command-name>", content_str)
            if match:
                skill_id = match.group(1)
                # Skip built-in commands
                if skill_id not in ("help", "clear", "resume", "compact", "config"):
                    invocations.append(SkillInvocation(
                        invocation_id=uuid,
                        skill_id=skill_id,
                        timestamp=timestamp,
                        message_index=idx,
                    ))

    return invocations


# ---------------------------------------------------------------------------
# Outcome classification
# ---------------------------------------------------------------------------


def _extract_user_text(entry: dict[str, Any]) -> str:
    """Extract text content from a user message entry."""
    msg = entry.get("message", {})
    content = msg.get("content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, dict) and block.get("type") == "text":
                parts.append(block.get("text", ""))
            elif isinstance(block, str):
                parts.append(block)
        return " ".join(parts)
    return ""


def _detect_revert(messages: list[dict[str, Any]], start_idx: int, end_idx: int) -> bool:
    """Check if a git revert command appears in the window."""
    for entry in messages[start_idx:end_idx]:
        if entry.get("type") != "assistant":
            continue
        content = entry.get("message", {}).get("content", [])
        if not isinstance(content, list):
            continue
        for block in content:
            if not isinstance(block, dict):
                continue
            if block.get("type") == "tool_use" and block.get("name") == "Bash":
                cmd = block.get("input", {}).get("command", "")
                if any(rc in cmd for rc in REVERT_COMMANDS):
                    return True
    return False


def _collect_ai_tools(messages: list[dict[str, Any]], start_idx: int, end_idx: int) -> list[str]:
    """Collect tool names used by the assistant in the window."""
    tools: set[str] = set()
    for entry in messages[start_idx:end_idx]:
        if entry.get("type") != "assistant":
            continue
        content = entry.get("message", {}).get("content", [])
        if not isinstance(content, list):
            continue
        for block in content:
            if isinstance(block, dict) and block.get("type") == "tool_use":
                name = block.get("name", "")
                if name:
                    tools.add(name)
    return sorted(tools)


def classify_outcome(
    messages: list[dict[str, Any]],
    invocation: SkillInvocation,
    next_invocation_idx: int | None = None,
) -> FeedbackEvent | None:
    """Classify user response after a skill invocation.

    Returns None for ambiguous outcomes (excluded from metrics).
    """
    start_idx = invocation.message_index + 1
    window_end = next_invocation_idx or len(messages)

    # Collect user messages within the influence window
    # Skip system-injected skill loading messages (contain "Base directory for this skill")
    SYSTEM_INJECT_MARKERS = ("Base directory for this skill", "<command-name>", "SKILL.md", "---\nname:")
    user_turns: list[tuple[int, dict[str, Any]]] = []
    for idx in range(start_idx, window_end):
        entry = messages[idx]
        if entry.get("type") == "user":
            text = _extract_user_text(entry)
            # Skip if this is a system-injected skill prompt, not real user input
            if any(marker in text for marker in SYSTEM_INJECT_MARKERS):
                continue
            user_turns.append((idx, entry))
            if len(user_turns) >= INFLUENCE_WINDOW:
                break

    if not user_turns:
        return None  # session ended, ambiguous

    # Check AI tool usage (if no tools used after skill load, exclude)
    ai_tools = _collect_ai_tools(messages, start_idx, window_end)

    # Classify based on first user response
    first_turn_idx, first_turn = user_turns[0]
    user_text = _extract_user_text(first_turn)
    lowered = user_text.lower()
    turns_to_feedback = 1

    # Check for revert in the window
    if _detect_revert(messages, start_idx, min(window_end, start_idx + 20)):
        return _build_event(
            invocation, "correction", 0.9, "revert", user_text,
            turns_to_feedback, ai_tools,
        )

    # Check correction signals across all user turns in window
    QUALIFIER_WORDS = ("但是", "不过", "但", "but", "however", "except", "though")
    for turn_num, (_, turn) in enumerate(user_turns, 1):
        text = _extract_user_text(turn)
        text_lower = text.lower()

        has_correction_kw = any(kw in text_lower for kw in CORRECTION_KEYWORDS_ZH + CORRECTION_KEYWORDS_EN)
        has_acceptance_kw = any(kw in text_lower for kw in ACCEPTANCE_KEYWORDS_ZH + ACCEPTANCE_KEYWORDS_EN)
        has_qualifier = any(q in text_lower for q in QUALIFIER_WORDS)

        # Partial: acceptance + qualifier ("可以，但是X要改") or correction + qualifier
        if has_qualifier and (has_acceptance_kw or has_correction_kw):
            return _build_event(
                invocation, "partial", 0.7, "partial", text,
                turn_num, ai_tools,
            )

        # Explicit rejection
        if has_correction_kw:
            return _build_event(
                invocation, "correction", 0.9, "rejection", text,
                turn_num, ai_tools,
            )

        # Redo request
        if any(kw in text_lower for kw in REDO_KEYWORDS):
            return _build_event(
                invocation, "correction", 0.9, "redo", text,
                turn_num, ai_tools,
            )

    # Check acceptance signals on first turn
    if any(kw in lowered for kw in ACCEPTANCE_KEYWORDS_ZH + ACCEPTANCE_KEYWORDS_EN):
        return _build_event(
            invocation, "acceptance", 0.8, None, user_text,
            turns_to_feedback, ai_tools,
        )

    # Silent continuation: user gives a new instruction (no correction of prior)
    # Heuristic: if the user message is long and doesn't reference the skill's output
    if len(user_text) > 20 and not any(kw in lowered for kw in ("?", "？")):
        return _build_event(
            invocation, "acceptance", 0.6, None, user_text,
            turns_to_feedback, ai_tools,
        )

    return None  # ambiguous


def _build_event(
    invocation: SkillInvocation,
    outcome: str,
    confidence: float,
    correction_type: str | None,
    user_text: str,
    turns_to_feedback: int,
    ai_tools: list[str],
) -> FeedbackEvent:
    event_id = hashlib.sha256(
        f"{invocation.invocation_id}:{invocation.skill_id}".encode()
    ).hexdigest()[:16]
    snippet = " ".join(user_text.split())[:200]
    dimension = attribute_dimension(snippet)
    return FeedbackEvent(
        event_id=event_id,
        timestamp=invocation.timestamp,
        session_id="",  # filled by caller
        skill_id=invocation.skill_id,
        invocation_uuid=invocation.invocation_id,
        outcome=outcome,
        confidence=confidence,
        correction_type=correction_type,
        user_message_snippet=snippet,
        turns_to_feedback=turns_to_feedback,
        ai_tools_used=ai_tools,
        dimension_hint=dimension,
    )


# ---------------------------------------------------------------------------
# Dimension attribution
# ---------------------------------------------------------------------------


def attribute_dimension(snippet: str) -> str | None:
    """Heuristic mapping of correction text to evaluator dimensions."""
    lowered = snippet.lower()
    for dimension, keywords in DIMENSION_KEYWORDS.items():
        if any(kw in lowered for kw in keywords):
            return dimension
    return None


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------


def write_feedback_jsonl(
    events: list[FeedbackEvent],
    output_path: Path,
    no_snippets: bool = False,
) -> Path:
    """Append feedback events to a JSONL file."""
    output_path.parent.mkdir(parents=True, exist_ok=True)

    # Read existing event IDs to avoid duplicates
    existing_ids: set[str] = set()
    if output_path.exists():
        with output_path.open("r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        existing_ids.add(json.loads(line).get("event_id", ""))
                    except json.JSONDecodeError:
                        continue

    new_count = 0
    with output_path.open("a", encoding="utf-8") as f:
        for event in events:
            if event.event_id in existing_ids:
                continue
            d = asdict(event)
            if no_snippets:
                d["user_message_snippet"] = ""
            f.write(json.dumps(d, ensure_ascii=False) + "\n")
            new_count += 1

    return output_path


def archive_old_events(feedback_path: Path, archive_dir: Path, days: int = 90) -> int:
    """Move events older than `days` to archive. Returns count archived."""
    if not feedback_path.exists():
        return 0

    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    keep: list[str] = []
    archive: list[str] = []

    with feedback_path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                entry = json.loads(line)
                ts = entry.get("timestamp", "")
                if ts:
                    entry_time = datetime.fromisoformat(ts.replace("Z", "+00:00"))
                    if entry_time < cutoff:
                        archive.append(line)
                        continue
            except (json.JSONDecodeError, ValueError):
                pass
            keep.append(line)

    if not archive:
        return 0

    archive_dir.mkdir(parents=True, exist_ok=True)
    archive_path = archive_dir / f"feedback-{cutoff.strftime('%Y%m%d')}.jsonl"
    with archive_path.open("a", encoding="utf-8") as f:
        for line in archive:
            f.write(line + "\n")

    with feedback_path.open("w", encoding="utf-8") as f:
        for line in keep:
            f.write(line + "\n")

    return len(archive)


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


def analyze_sessions(
    session_dir: Path,
    skill_filter: str | None = None,
) -> list[FeedbackEvent]:
    """Analyze all sessions and return feedback events."""
    all_events: list[FeedbackEvent] = []

    for session_path in iter_session_files(session_dir):
        session_id = extract_session_id(session_path)
        project = extract_project(session_path)
        messages = parse_session(session_path)

        if not messages:
            continue

        invocations = detect_skill_invocations(messages)

        if skill_filter:
            invocations = [inv for inv in invocations if inv.skill_id == skill_filter]

        for i, invocation in enumerate(invocations):
            next_idx = invocations[i + 1].message_index if i + 1 < len(invocations) else None
            event = classify_outcome(messages, invocation, next_idx)
            if event:
                event.session_id = session_id
                all_events.append(event)

    return all_events


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    session_dir = Path(args.session_dir).expanduser()

    # Check opt-out
    opt_out = Path.home() / ".claude" / "feedback-config.json"
    if opt_out.exists():
        try:
            config = json.loads(opt_out.read_text())
            if not config.get("enabled", True):
                print("Feedback collection disabled via feedback-config.json")
                return 0
        except (json.JSONDecodeError, OSError):
            pass

    if not session_dir.exists():
        print(f"Session directory not found: {session_dir}", file=sys.stderr)
        return 1

    events = analyze_sessions(session_dir, skill_filter=args.skill_filter)

    output_path = Path(args.output)
    write_feedback_jsonl(events, output_path, no_snippets=args.no_snippets)

    # Archive old events
    archive_dir = output_path.parent / "archive"
    archived = archive_old_events(output_path, archive_dir)

    # Print summary
    from collections import Counter
    outcomes = Counter(e.outcome for e in events)
    skills = Counter(e.skill_id for e in events)
    print(f"Analyzed {len(events)} feedback events")
    print(f"  Outcomes: {dict(outcomes)}")
    print(f"  Top skills: {skills.most_common(5)}")
    if archived:
        print(f"  Archived {archived} old events")
    print(str(output_path))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Metrics computation for session feedback data.

Computes correction_rate, correction_trend, and dimension hotspots
from feedback.jsonl event data.
"""

from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


def load_feedback_events(path: Path) -> list[dict[str, Any]]:
    """Load feedback events from a JSONL file."""
    events: list[dict[str, Any]] = []
    if not path.exists():
        return events
    with path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                events.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return events


def filter_by_skill(events: list[dict[str, Any]], skill_id: str) -> list[dict[str, Any]]:
    """Filter events to a specific skill."""
    return [e for e in events if e.get("skill_id") == skill_id]


def compute_correction_rate(events: list[dict[str, Any]], skill_id: str) -> dict[str, Any]:
    """Compute correction rate for a skill.

    Formula: (corrections + 0.5 * partials) / total_invocations
    Returns insufficient_data when sample_size < 5.
    """
    skill_events = filter_by_skill(events, skill_id)

    corrections = sum(1 for e in skill_events if e.get("outcome") == "correction")
    partials = sum(1 for e in skill_events if e.get("outcome") == "partial")
    acceptances = sum(1 for e in skill_events if e.get("outcome") == "acceptance")
    total = corrections + partials + acceptances

    if total == 0:
        return {
            "skill_id": skill_id,
            "correction_rate": 0.0,
            "sample_size": 0,
            "sufficient_data": False,
            "corrections": 0,
            "partials": 0,
            "acceptances": 0,
        }

    rate = (corrections + 0.5 * partials) / total

    return {
        "skill_id": skill_id,
        "correction_rate": round(rate, 4),
        "sample_size": total,
        "sufficient_data": total >= 5,
        "corrections": corrections,
        "partials": partials,
        "acceptances": acceptances,
    }


def _parse_timestamp(ts: str) -> datetime | None:
    """Parse an ISO timestamp string."""
    try:
        return datetime.fromisoformat(ts.replace("Z", "+00:00"))
    except (ValueError, AttributeError):
        return None


def compute_correction_trend(
    events: list[dict[str, Any]],
    skill_id: str,
    window_days: int = 30,
) -> dict[str, Any]:
    """Compute correction rate trend over time.

    Returns: correction_rate(last window) - correction_rate(prior window).
    Positive = getting worse, Negative = improving.
    """
    now = datetime.now(timezone.utc)
    cutoff_recent = now - timedelta(days=window_days)
    cutoff_prior = now - timedelta(days=window_days * 2)

    skill_events = filter_by_skill(events, skill_id)

    recent: list[dict[str, Any]] = []
    prior: list[dict[str, Any]] = []

    for e in skill_events:
        ts = _parse_timestamp(e.get("timestamp", ""))
        if ts is None:
            continue
        if ts >= cutoff_recent:
            recent.append(e)
        elif ts >= cutoff_prior:
            prior.append(e)

    recent_rate = compute_correction_rate(recent, skill_id)
    prior_rate = compute_correction_rate(prior, skill_id)

    trend = recent_rate["correction_rate"] - prior_rate["correction_rate"]

    return {
        "skill_id": skill_id,
        "trend": round(trend, 4),
        "recent_rate": recent_rate["correction_rate"],
        "prior_rate": prior_rate["correction_rate"],
        "recent_sample": recent_rate["sample_size"],
        "prior_sample": prior_rate["sample_size"],
        "direction": "worsening" if trend > 0.05 else "improving" if trend < -0.05 else "stable",
    }


def compute_hotspot_dimensions(
    events: list[dict[str, Any]],
    skill_id: str,
) -> dict[str, int]:
    """Group corrections by dimension_hint, return frequency map."""
    skill_events = filter_by_skill(events, skill_id)
    hotspots: dict[str, int] = {}
    for e in skill_events:
        if e.get("outcome") not in ("correction", "partial"):
            continue
        dim = e.get("dimension_hint")
        if dim:
            hotspots[dim] = hotspots.get(dim, 0) + 1
    return hotspots


def compute_all_skill_metrics(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Compute metrics for all skills in the event set."""
    skill_ids = {e.get("skill_id", "") for e in events if e.get("skill_id")}
    results = []
    for skill_id in sorted(skill_ids):
        rate = compute_correction_rate(events, skill_id)
        hotspots = compute_hotspot_dimensions(events, skill_id)
        rate["hotspot_dimensions"] = hotspots
        results.append(rate)
    return results


def format_metrics_report(metrics: list[dict[str, Any]]) -> str:
    """Format metrics as a human-readable report."""
    lines = ["Skill Feedback Metrics", "=" * 40]
    for m in sorted(metrics, key=lambda x: -x.get("correction_rate", 0)):
        suffix = "" if m["sufficient_data"] else " (insufficient data)"
        lines.append(
            f"  {m['skill_id']}: correction_rate={m['correction_rate']:.2f} "
            f"(n={m['sample_size']}, "
            f"corrections={m['corrections']}, "
            f"partials={m['partials']}, "
            f"acceptances={m['acceptances']}){suffix}"
        )
        hotspots = m.get("hotspot_dimensions", {})
        if hotspots:
            top = sorted(hotspots.items(), key=lambda x: -x[1])[:3]
            lines.append(f"    hotspots: {', '.join(f'{d}={c}' for d, c in top)}")
    return "\n".join(lines)

"""Tests for session feedback analyzer."""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

_SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
if str(_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS))

import analyze  # noqa: E402


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _skill_tool_use(uuid: str, skill_id: str, ts: str = "2026-04-05T10:00:00Z"):
    return {
        "type": "assistant",
        "uuid": uuid,
        "timestamp": ts,
        "message": {
            "content": [{
                "type": "tool_use",
                "name": "Skill",
                "input": {"skill": skill_id},
                "id": f"toolu_{uuid[:8]}",
            }],
        },
    }


def _slash_command(uuid: str, command: str, ts: str = "2026-04-05T10:00:00Z"):
    return {
        "type": "system",
        "subtype": "local_command",
        "uuid": uuid,
        "timestamp": ts,
        "content": f"<command-name>/{command}</command-name>\n<command-message>{command}</command-message>",
    }


def _user_msg(uuid: str, text: str, parent: str = "", ts: str = "2026-04-05T10:01:00Z"):
    return {
        "type": "user",
        "uuid": uuid,
        "parentUuid": parent,
        "timestamp": ts,
        "message": {"role": "user", "content": text},
    }


def _assistant_msg(uuid: str, text: str = "", tools: list[dict] | None = None, ts: str = "2026-04-05T10:00:30Z"):
    content = []
    if text:
        content.append({"type": "text", "text": text})
    for t in (tools or []):
        content.append(t)
    return {
        "type": "assistant",
        "uuid": uuid,
        "timestamp": ts,
        "message": {"content": content},
    }


def _bash_tool_use(command: str):
    return {
        "type": "tool_use",
        "name": "Bash",
        "input": {"command": command},
        "id": "toolu_bash",
    }


# ---------------------------------------------------------------------------
# Tests: detect_skill_invocations
# ---------------------------------------------------------------------------


class TestDetectSkillInvocations:
    def test_detects_tool_use_skill(self):
        messages = [
            _skill_tool_use("uuid-1", "cpp-expert"),
        ]
        result = analyze.detect_skill_invocations(messages)
        assert len(result) == 1
        assert result[0].skill_id == "cpp-expert"
        assert result[0].invocation_id == "uuid-1"
        assert result[0].message_index == 0

    def test_detects_slash_command(self):
        messages = [
            _slash_command("uuid-2", "deslop"),
        ]
        result = analyze.detect_skill_invocations(messages)
        assert len(result) == 1
        assert result[0].skill_id == "deslop"

    def test_skips_builtin_commands(self):
        messages = [
            _slash_command("uuid-3", "help"),
            _slash_command("uuid-4", "clear"),
            _slash_command("uuid-5", "resume"),
        ]
        result = analyze.detect_skill_invocations(messages)
        assert len(result) == 0

    def test_detects_multiple_invocations(self):
        messages = [
            _skill_tool_use("uuid-1", "cpp-expert"),
            _user_msg("uuid-u1", "ok"),
            _slash_command("uuid-2", "deslop"),
        ]
        result = analyze.detect_skill_invocations(messages)
        assert len(result) == 2
        assert result[0].skill_id == "cpp-expert"
        assert result[1].skill_id == "deslop"

    def test_ignores_non_skill_tool_use(self):
        messages = [{
            "type": "assistant",
            "uuid": "uuid-x",
            "timestamp": "2026-04-05T10:00:00Z",
            "message": {
                "content": [{
                    "type": "tool_use",
                    "name": "Read",
                    "input": {"file_path": "/tmp/foo"},
                    "id": "toolu_read",
                }],
            },
        }]
        result = analyze.detect_skill_invocations(messages)
        assert len(result) == 0


# ---------------------------------------------------------------------------
# Tests: classify_outcome
# ---------------------------------------------------------------------------


class TestClassifyOutcome:
    def test_correction_rejection_zh(self):
        messages = [
            _skill_tool_use("inv-1", "cpp-expert"),
            _user_msg("u-1", "不对，应该用snake_case"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "correction"
        assert event.correction_type == "rejection"
        assert event.confidence >= 0.9

    def test_correction_rejection_en(self):
        messages = [
            _skill_tool_use("inv-1", "code-review"),
            _user_msg("u-1", "No, that's not right. Use the other approach"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "correction"

    def test_correction_redo(self):
        messages = [
            _skill_tool_use("inv-1", "deslop"),
            _user_msg("u-1", "重新来，这个效果不好"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "correction"
        assert event.correction_type in ("redo", "rejection")

    def test_correction_revert(self):
        messages = [
            _skill_tool_use("inv-1", "code-review"),
            _assistant_msg("a-1", tools=[_bash_tool_use("git checkout -- src/main.py")]),
            _user_msg("u-1", "just undo that"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "correction"
        assert event.correction_type == "revert"

    def test_partial_correction(self):
        messages = [
            _skill_tool_use("inv-1", "cpp-expert"),
            _user_msg("u-1", "这个可以，但是命名应该用camelCase"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "partial"

    def test_acceptance_explicit(self):
        messages = [
            _skill_tool_use("inv-1", "deslop"),
            _user_msg("u-1", "looks good, thanks"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "acceptance"

    def test_acceptance_silent_continuation(self):
        messages = [
            _skill_tool_use("inv-1", "cpp-expert"),
            _user_msg("u-1", "Now let's work on the authentication module next"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert event.outcome == "acceptance"
        assert event.confidence <= 0.7

    def test_ambiguous_no_user_response(self):
        messages = [
            _skill_tool_use("inv-1", "cpp-expert"),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is None

    def test_window_bounded_by_next_invocation(self):
        messages = [
            _skill_tool_use("inv-1", "cpp-expert", ts="2026-04-05T10:00:00Z"),
            _user_msg("u-1", "looks good, thanks for the help"),
            _skill_tool_use("inv-2", "deslop", ts="2026-04-05T10:01:00Z"),
            _user_msg("u-2", "不对"),
        ]
        invocations = analyze.detect_skill_invocations(messages)
        # First invocation's window should NOT include "不对" after second invocation
        event1 = analyze.classify_outcome(messages, invocations[0], invocations[1].message_index)
        assert event1 is not None
        assert event1.outcome == "acceptance"

    def test_snippet_truncation(self):
        messages = [
            _skill_tool_use("inv-1", "cpp-expert"),
            _user_msg("u-1", "不对 " + "x" * 300),
        ]
        inv = analyze.detect_skill_invocations(messages)[0]
        event = analyze.classify_outcome(messages, inv)
        assert event is not None
        assert len(event.user_message_snippet) <= 200


# ---------------------------------------------------------------------------
# Tests: attribute_dimension
# ---------------------------------------------------------------------------


class TestAttributeDimension:
    def test_accuracy(self):
        assert analyze.attribute_dimension("命名不对") == "accuracy"
        assert analyze.attribute_dimension("wrong naming convention") == "accuracy"

    def test_coverage(self):
        assert analyze.attribute_dimension("你漏了错误处理") == "coverage"
        assert analyze.attribute_dimension("missing error handling") == "coverage"

    def test_security(self):
        assert analyze.attribute_dimension("token exposed") == "security"

    def test_none_for_generic(self):
        assert analyze.attribute_dimension("hello world") is None


# ---------------------------------------------------------------------------
# Tests: write_feedback_jsonl
# ---------------------------------------------------------------------------


class TestWriteFeedbackJsonl:
    def test_creates_file(self, tmp_path):
        event = analyze.FeedbackEvent(
            event_id="abc123",
            timestamp="2026-04-05T10:00:00Z",
            session_id="sess-1",
            skill_id="cpp-expert",
            invocation_uuid="inv-1",
            outcome="correction",
            confidence=0.9,
            correction_type="rejection",
            user_message_snippet="不对",
            turns_to_feedback=1,
            ai_tools_used=["Read", "Edit"],
            dimension_hint="accuracy",
        )
        out = tmp_path / "feedback.jsonl"
        analyze.write_feedback_jsonl([event], out)
        assert out.exists()
        lines = out.read_text().strip().split("\n")
        assert len(lines) == 1
        parsed = json.loads(lines[0])
        assert parsed["event_id"] == "abc123"
        assert parsed["outcome"] == "correction"

    def test_deduplicates(self, tmp_path):
        event = analyze.FeedbackEvent(
            event_id="abc123", timestamp="", session_id="", skill_id="x",
            invocation_uuid="", outcome="correction", confidence=0.9,
            correction_type=None, user_message_snippet="", turns_to_feedback=1,
            ai_tools_used=[], dimension_hint=None,
        )
        out = tmp_path / "feedback.jsonl"
        analyze.write_feedback_jsonl([event], out)
        analyze.write_feedback_jsonl([event], out)  # write again
        lines = out.read_text().strip().split("\n")
        assert len(lines) == 1  # no duplicate

    def test_no_snippets_mode(self, tmp_path):
        event = analyze.FeedbackEvent(
            event_id="def456", timestamp="", session_id="", skill_id="x",
            invocation_uuid="", outcome="correction", confidence=0.9,
            correction_type=None, user_message_snippet="sensitive text",
            turns_to_feedback=1, ai_tools_used=[], dimension_hint=None,
        )
        out = tmp_path / "feedback.jsonl"
        analyze.write_feedback_jsonl([event], out, no_snippets=True)
        parsed = json.loads(out.read_text().strip())
        assert parsed["user_message_snippet"] == ""


# ---------------------------------------------------------------------------
# Tests: archive_old_events
# ---------------------------------------------------------------------------


class TestArchiveOldEvents:
    def test_archives_old_events(self, tmp_path):
        feedback = tmp_path / "feedback.jsonl"
        archive_dir = tmp_path / "archive"

        old_event = {"event_id": "old", "timestamp": "2025-01-01T00:00:00Z", "outcome": "correction"}
        new_event = {"event_id": "new", "timestamp": "2026-04-05T00:00:00Z", "outcome": "acceptance"}

        with feedback.open("w") as f:
            f.write(json.dumps(old_event) + "\n")
            f.write(json.dumps(new_event) + "\n")

        count = analyze.archive_old_events(feedback, archive_dir, days=90)
        assert count == 1

        remaining = feedback.read_text().strip().split("\n")
        assert len(remaining) == 1
        assert json.loads(remaining[0])["event_id"] == "new"

        archived_files = list(archive_dir.glob("*.jsonl"))
        assert len(archived_files) == 1

"""Tests for session feedback metrics."""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

_SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
if str(_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS))

import metrics  # noqa: E402


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _event(skill_id: str, outcome: str, ts: str = "2026-04-05T10:00:00Z", dim: str | None = None):
    return {
        "event_id": f"{skill_id}-{outcome}-{ts}",
        "timestamp": ts,
        "session_id": "sess-1",
        "skill_id": skill_id,
        "outcome": outcome,
        "confidence": 0.9,
        "correction_type": "rejection" if outcome == "correction" else None,
        "dimension_hint": dim,
    }


# ---------------------------------------------------------------------------
# Tests: compute_correction_rate
# ---------------------------------------------------------------------------


class TestCorrectionRate:
    def test_basic(self):
        events = [
            _event("skill-a", "correction"),
            _event("skill-a", "correction"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
        ]
        result = metrics.compute_correction_rate(events, "skill-a")
        assert result["correction_rate"] == 0.4  # 2/5
        assert result["sample_size"] == 5
        assert result["sufficient_data"] is True

    def test_with_partials(self):
        events = [
            _event("skill-a", "correction"),
            _event("skill-a", "partial"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
        ]
        result = metrics.compute_correction_rate(events, "skill-a")
        # (1 + 0.5*1) / 5 = 0.3
        assert result["correction_rate"] == 0.3

    def test_all_acceptances(self):
        events = [_event("skill-a", "acceptance") for _ in range(6)]
        result = metrics.compute_correction_rate(events, "skill-a")
        assert result["correction_rate"] == 0.0

    def test_all_corrections(self):
        events = [_event("skill-a", "correction") for _ in range(5)]
        result = metrics.compute_correction_rate(events, "skill-a")
        assert result["correction_rate"] == 1.0

    def test_insufficient_data(self):
        events = [
            _event("skill-a", "correction"),
            _event("skill-a", "acceptance"),
        ]
        result = metrics.compute_correction_rate(events, "skill-a")
        assert result["sufficient_data"] is False

    def test_empty(self):
        result = metrics.compute_correction_rate([], "skill-a")
        assert result["correction_rate"] == 0.0
        assert result["sample_size"] == 0
        assert result["sufficient_data"] is False

    def test_filters_by_skill(self):
        events = [
            _event("skill-a", "correction"),
            _event("skill-b", "acceptance"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
            _event("skill-a", "acceptance"),
        ]
        result = metrics.compute_correction_rate(events, "skill-a")
        assert result["correction_rate"] == 0.2  # 1/5
        assert result["sample_size"] == 5


# ---------------------------------------------------------------------------
# Tests: compute_correction_trend
# ---------------------------------------------------------------------------


class TestCorrectionTrend:
    def test_stable(self):
        events = [_event("s", "acceptance", ts="2026-04-04T10:00:00Z") for _ in range(5)]
        events += [_event("s", "acceptance", ts="2026-03-04T10:00:00Z") for _ in range(5)]
        result = metrics.compute_correction_trend(events, "s", window_days=30)
        assert result["direction"] == "stable"

    def test_no_events(self):
        result = metrics.compute_correction_trend([], "s")
        assert result["trend"] == 0.0


# ---------------------------------------------------------------------------
# Tests: compute_hotspot_dimensions
# ---------------------------------------------------------------------------


class TestHotspotDimensions:
    def test_groups_by_dimension(self):
        events = [
            _event("s", "correction", dim="accuracy"),
            _event("s", "correction", dim="accuracy"),
            _event("s", "correction", dim="coverage"),
            _event("s", "acceptance"),  # no dimension, not a correction
        ]
        result = metrics.compute_hotspot_dimensions(events, "s")
        assert result == {"accuracy": 2, "coverage": 1}

    def test_ignores_none_dimension(self):
        events = [
            _event("s", "correction", dim=None),
        ]
        result = metrics.compute_hotspot_dimensions(events, "s")
        assert result == {}

    def test_includes_partials(self):
        events = [
            _event("s", "partial", dim="efficiency"),
        ]
        result = metrics.compute_hotspot_dimensions(events, "s")
        assert result == {"efficiency": 1}


# ---------------------------------------------------------------------------
# Tests: format_metrics_report
# ---------------------------------------------------------------------------


class TestFormatReport:
    def test_basic_format(self):
        metrics_data = [{
            "skill_id": "cpp-expert",
            "correction_rate": 0.4,
            "sample_size": 10,
            "sufficient_data": True,
            "corrections": 3,
            "partials": 2,
            "acceptances": 5,
            "hotspot_dimensions": {"accuracy": 2, "coverage": 1},
        }]
        report = metrics.format_metrics_report(metrics_data)
        assert "cpp-expert" in report
        assert "0.40" in report
        assert "accuracy=2" in report