Retrospective

{
  "name": "retrospective",
  "description": "retrospective",
  "author": {
    "name": "Gleb Kalinin"
  },
  "repository": "https://github.com/glebis/claude-skills",
  "license": "MIT"
}

"""Retrospective engine — deterministic logic for the /retrospective skill.

Extracts testable functions: gate check, finding generation, candidate ranking,
dedup against existing memories, and conflict detection against skill files.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum


class Mode(Enum):
    FAST = "fast"
    FULL = "full"
    SKIP = "skip"


class CandidateType(Enum):
    SKILL_UPDATE = "skill_update"
    FEEDBACK_MEMORY = "feedback_memory"
    CLAUDE_MD_RULE = "claude_md_rule"
    LINEAR_TASK = "linear_task"


class SignalSource(Enum):
    USER_CORRECTION = "user_correction"
    FAILED_SKILL = "failed_skill"
    REPEATED_PATTERN = "repeated_pattern"
    ERROR_PATTERN = "error_pattern"
    WORKFLOW_PATTERN = "workflow_pattern"


SIGNAL_PRIORITY = {
    SignalSource.USER_CORRECTION: 1,
    SignalSource.FAILED_SKILL: 2,
    SignalSource.REPEATED_PATTERN: 3,
    SignalSource.ERROR_PATTERN: 4,
    SignalSource.WORKFLOW_PATTERN: 5,
}

TOOL_CALL_THRESHOLD = 8
MAX_CANDIDATES = 5


@dataclass
class SkillInvocation:
    name: str
    status: str  # "success" | "failed"
    error: str | None = None


@dataclass
class Memory:
    name: str
    content: str


@dataclass
class SessionData:
    tool_calls: int
    skills_invoked: list[SkillInvocation] = field(default_factory=list)
    user_corrections: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)
    repeated_patterns: list[str] = field(default_factory=list)
    existing_memories: list[Memory] = field(default_factory=list)
    existing_skill_content: dict[str, str] = field(default_factory=dict)
    linear_configured: bool = True


@dataclass
class Candidate:
    type: CandidateType
    source: SignalSource
    label: str
    description: str
    target_skill: str | None = None
    proposed_content: str = ""
    conflict: bool = False


def gate_check(tool_calls: int) -> Mode:
    if tool_calls < TOOL_CALL_THRESHOLD:
        return Mode.FAST
    return Mode.FULL


def _normalize(text: str) -> str:
    return text.lower().strip()


def _content_overlaps(needle: str, haystack: str, threshold: float = 0.5) -> bool:
    """Check if key terms from needle appear in haystack."""
    needle_words = set(_normalize(needle).split())
    haystack_lower = _normalize(haystack)
    stopwords = {"the", "a", "an", "is", "was", "to", "in", "for", "of", "and", "or", "not", "don't", "do", "use"}
    meaningful = needle_words - stopwords
    if not meaningful:
        return False
    matches = sum(1 for w in meaningful if w in haystack_lower)
    return (matches / len(meaningful)) >= threshold


def dedup_against_memory(
    candidates: list[Candidate],
    existing_memories: list[Memory],
) -> list[Candidate]:
    result = []
    for c in candidates:
        is_dup = False
        for mem in existing_memories:
            if _content_overlaps(c.proposed_content or c.description, mem.content):
                is_dup = True
                break
        if not is_dup:
            result.append(c)
    return result


def check_skill_conflict(
    proposed_content: str,
    existing_content: str,
) -> bool:
    """Return True if proposed content overlaps with existing skill file content."""
    return _content_overlaps(proposed_content, existing_content)


def generate_findings(session: SessionData) -> list[Candidate]:
    candidates: list[Candidate] = []

    for correction in session.user_corrections:
        candidates.append(Candidate(
            type=CandidateType.FEEDBACK_MEMORY,
            source=SignalSource.USER_CORRECTION,
            label=f"Remember: {correction[:60]}",
            description=correction,
            proposed_content=correction,
        ))

    for skill in session.skills_invoked:
        if skill.status == "failed" and skill.error:
            candidates.append(Candidate(
                type=CandidateType.SKILL_UPDATE,
                source=SignalSource.FAILED_SKILL,
                label=f"Update {skill.name} skill",
                description=f"{skill.name} failed: {skill.error}",
                target_skill=skill.name,
                proposed_content=skill.error,
            ))

    for pattern in session.repeated_patterns:
        if any(s.name for s in session.skills_invoked if s.status == "success") and len(session.skills_invoked) >= 3:
            candidates.append(Candidate(
                type=CandidateType.FEEDBACK_MEMORY,
                source=SignalSource.WORKFLOW_PATTERN,
                label=f"Workflow: {pattern[:60]}",
                description=pattern,
                proposed_content=pattern,
            ))
        else:
            candidates.append(Candidate(
                type=CandidateType.FEEDBACK_MEMORY,
                source=SignalSource.REPEATED_PATTERN,
                label=f"Pattern: {pattern[:60]}",
                description=pattern,
                proposed_content=pattern,
            ))

    return candidates


def filter_by_skill_content(
    candidates: list[Candidate],
    existing_skill_content: dict[str, str],
) -> list[Candidate]:
    result = []
    for c in candidates:
        if c.type == CandidateType.SKILL_UPDATE and c.target_skill:
            content = existing_skill_content.get(c.target_skill, "")
            if content and check_skill_conflict(c.proposed_content, content):
                continue
        result.append(c)
    return result


def filter_linear_candidates(
    candidates: list[Candidate],
    linear_configured: bool,
) -> list[Candidate]:
    if linear_configured:
        return candidates
    return [c for c in candidates if c.type != CandidateType.LINEAR_TASK]


def rank_candidates(candidates: list[Candidate]) -> list[Candidate]:
    return sorted(candidates, key=lambda c: SIGNAL_PRIORITY.get(c.source, 99))


def process_session(session: SessionData) -> tuple[Mode, list[Candidate]]:
    """Full pipeline: gate → generate → dedup → filter → rank → cap."""
    mode = gate_check(session.tool_calls)
    if mode == Mode.FAST:
        return mode, []

    candidates = generate_findings(session)
    candidates = dedup_against_memory(candidates, session.existing_memories)
    candidates = filter_by_skill_content(candidates, session.existing_skill_content)
    candidates = filter_linear_candidates(candidates, session.linear_configured)
    candidates = rank_candidates(candidates)
    candidates = candidates[:MAX_CANDIDATES]

    return mode, candidates

name: productive-long-session
description: >
  90-min session: 3 skills invoked, telegram failed, user corrected Claude twice.
  Should trigger full mode with skill update and feedback memory candidates.

mock_session:
  tool_calls: 47
  skills_invoked:
    - name: linear
      status: success
    - name: telegram
      status: failed
      error: "Telethon session file not found"
    - name: de-ai
      status: success
  user_corrections:
    - "no, use Telethon directly for private DMs"
    - "don't mock the database in integration tests"
  errors:
    - "Telethon session file not found"
    - "FileNotFoundError: user.session"
  repeated_patterns: []
  existing_memories:
    - name: "feedback_testing"
      content: "Integration tests must hit a real database"

expected:
  mode: full
  candidate_count:
    min: 2
    max: 5
  must_include:
    - type: skill_update
      target_skill: telegram
      contains: "Telethon"
    - type: feedback_memory
      contains: "Telethon"
  must_not_include:
    - type: feedback_memory
      contains: "don't mock the database"
      reason: "already exists in memory"
  skip_triggered: false

name: short-session
description: >
  Quick 3-tool session: user asked one question, got an answer, left.
  Should trigger fast mode with single optional question.

mock_session:
  tool_calls: 3
  skills_invoked: []
  user_corrections: []
  errors: []
  repeated_patterns: []
  existing_memories: []

expected:
  mode: fast
  candidate_count:
    min: 0
    max: 1
  must_include: []
  must_not_include: []
  skip_triggered: false

name: rough-session
description: >
  User hit multiple errors, expressed frustration. Should offer skip path
  immediately and not force the full flow.

mock_session:
  tool_calls: 25
  skills_invoked:
    - name: deploy
      status: failed
      error: "Build failed: TypeScript errors"
    - name: tdd
      status: failed
      error: "Test suite timeout after 120s"
  user_corrections:
    - "this is completely wrong, start over"
  errors:
    - "Build failed: TypeScript errors"
    - "Test suite timeout after 120s"
    - "ECONNREFUSED localhost:3000"
    - "Permission denied: .env.local"
  repeated_patterns:
    - "retried same failing command 3 times"
  existing_memories: []

expected:
  mode: full
  candidate_count:
    min: 1
    max: 5
  must_include:
    - type: skill_update
      target_skill: deploy
      contains: "TypeScript"
  must_not_include: []
  skip_triggered: false
  rough_session_handling: true

name: no-skills-used
description: >
  Medium session with pure code editing, no skills invoked.
  Should only surface CLAUDE.md and memory candidates, no skill updates.

mock_session:
  tool_calls: 18
  skills_invoked: []
  user_corrections:
    - "always use single quotes in this project"
  errors:
    - "ESLint: prefer-single-quotes"
  repeated_patterns:
    - "fixed quote style 4 times across different files"
  existing_memories: []

expected:
  mode: full
  candidate_count:
    min: 1
    max: 3
  must_include:
    - type: claude_md_rule
      contains: "single quotes"
  must_not_include:
    - type: skill_update
      reason: "no skills were used"
  skip_triggered: false

name: many-learnings
description: >
  Long session with 12+ distinct findings. Should cap candidates at 5,
  ranked by signal strength (corrections > failures > patterns).

mock_session:
  tool_calls: 85
  skills_invoked:
    - name: telegram
      status: failed
      error: "send_message: chat type mismatch"
    - name: linear
      status: success
    - name: de-ai
      status: success
    - name: presentation-generator
      status: success
    - name: gpt-image-2
      status: failed
      error: "rate limit exceeded"
  user_corrections:
    - "use Telethon directly for DMs"
    - "don't use /api/generate, use /api/chat for Qwen"
    - "always check .trash before linking"
  errors:
    - "send_message: chat type mismatch"
    - "rate limit exceeded"
    - "Ollama: empty content on long prompts"
    - "ENOMEM: layers fight for RAM"
    - "pandoc: missing LaTeX package"
    - "ESLint warnings in 6 files"
  repeated_patterns:
    - "retried gpt-image-2 after rate limit 3 times"
    - "switched from Qwen3 to Qwen2.5 after thinking mode issues"
  existing_memories:
    - name: "feedback_telethon"
      content: "use Telethon directly for DMs"

expected:
  mode: full
  candidate_count:
    min: 5
    max: 5
  must_include:
    - type: skill_update
      target_skill: telegram
      contains: "chat type"
    - type: feedback_memory
      contains: "Qwen"
  must_not_include:
    - type: feedback_memory
      contains: "Telethon directly for DMs"
      reason: "already exists in memory"
  skip_triggered: false
  ranking_order:
    - "user_corrections first"
    - "failed_skills second"
    - "repeated_patterns third"

name: duplicate-memory
description: >
  Session where the main finding already exists in memory.
  Dedup should filter it out, leaving fewer or zero candidates.

mock_session:
  tool_calls: 20
  skills_invoked:
    - name: telegram
      status: failed
      error: "chat type mismatch for private DMs"
  user_corrections:
    - "use Telethon directly for private DMs"
  errors:
    - "chat type mismatch for private DMs"
  repeated_patterns: []
  existing_memories:
    - name: "feedback_telethon_dms"
      content: "Use Telethon client.send_message directly for private DMs — telegram skill misclassifies some private chats"
    - name: "feedback_telegram_chat_type"
      content: "telegram skill send command misclassifies some private chats as channels due to get_chat_type() bug"

expected:
  mode: full
  candidate_count:
    min: 0
    max: 2
  must_include: []
  must_not_include:
    - type: feedback_memory
      contains: "Telethon directly"
      reason: "already exists in feedback_telethon_dms"
    - type: feedback_memory
      contains: "chat type"
      reason: "already exists in feedback_telegram_chat_type"
  skip_triggered: false

name: skill-file-conflict
description: >
  Session finding contradicts existing content in a skill file.
  Engine should detect the conflict and flag it instead of writing.

mock_session:
  tool_calls: 30
  skills_invoked:
    - name: pdf-generation
      status: success
  user_corrections: []
  errors:
    - "pandoc: missing babel-lang for Russian"
  repeated_patterns: []
  existing_memories: []
  existing_skill_content:
    pdf-generation: |
      ## Known Issues
      ### 2025-01-01
      - Russian text needs `babel-lang: russian` in YAML frontmatter

expected:
  mode: full
  candidate_count:
    min: 0
    max: 3
  must_include: []
  must_not_include:
    - type: skill_update
      target_skill: pdf-generation
      contains: "babel-lang"
      reason: "already documented in skill file"
  skip_triggered: false

name: linear-not-configured
description: >
  Session where a bug worth tracking was found, but Linear CLI is not configured.
  Linear candidate should be omitted gracefully.

mock_session:
  tool_calls: 22
  skills_invoked:
    - name: telegram
      status: failed
      error: "send_message fails for group chats"
  user_corrections: []
  errors:
    - "send_message fails for group chats"
  repeated_patterns: []
  existing_memories: []
  linear_configured: false

expected:
  mode: full
  candidate_count:
    min: 1
    max: 4
  must_include:
    - type: skill_update
      target_skill: telegram
      contains: "group chats"
  must_not_include:
    - type: linear_task
      reason: "linear CLI not configured"
  skip_triggered: false

name: cross-skill-workflow
description: >
  Session chaining multiple skills in a pipeline.
  Should detect the workflow pattern as a candidate.

mock_session:
  tool_calls: 60
  skills_invoked:
    - name: telegram
      status: success
    - name: linear
      status: success
    - name: de-ai
      status: success
    - name: presentation-generator
      status: success
  user_corrections: []
  errors: []
  repeated_patterns:
    - "telegram output fed into linear task creation"
    - "presentation content humanized via de-ai"
  existing_memories: []

expected:
  mode: full
  candidate_count:
    min: 1
    max: 5
  must_include:
    - type: feedback_memory
      contains: "workflow"
  must_not_include: []
  skip_triggered: false

"""Tests for retro_engine — driven by YAML scenario fixtures."""

from pathlib import Path

import yaml
import pytest

from retro_engine import (
    CandidateType,
    Memory,
    Mode,
    SessionData,
    SignalSource,
    SkillInvocation,
    check_skill_conflict,
    dedup_against_memory,
    filter_by_skill_content,
    filter_linear_candidates,
    gate_check,
    generate_findings,
    process_session,
    rank_candidates,
    MAX_CANDIDATES,
)

SCENARIOS_DIR = Path(__file__).parent / "scenarios"


def load_scenario(name: str) -> dict:
    path = SCENARIOS_DIR / name
    return yaml.safe_load(path.read_text())


def build_session(scenario: dict) -> SessionData:
    ms = scenario["mock_session"]
    return SessionData(
        tool_calls=ms["tool_calls"],
        skills_invoked=[
            SkillInvocation(name=s["name"], status=s["status"], error=s.get("error"))
            for s in ms.get("skills_invoked", [])
        ],
        user_corrections=ms.get("user_corrections", []),
        errors=ms.get("errors", []),
        repeated_patterns=ms.get("repeated_patterns", []),
        existing_memories=[
            Memory(name=m["name"], content=m["content"])
            for m in ms.get("existing_memories", [])
        ],
        existing_skill_content=ms.get("existing_skill_content", {}),
        linear_configured=ms.get("linear_configured", True),
    )


# ── Gate check tests ──


class TestGateCheck:
    def test_short_session_returns_fast(self):
        assert gate_check(3) == Mode.FAST

    def test_threshold_returns_full(self):
        assert gate_check(8) == Mode.FULL

    def test_long_session_returns_full(self):
        assert gate_check(85) == Mode.FULL

    def test_boundary_below_threshold(self):
        assert gate_check(7) == Mode.FAST


# ── Scenario: productive long session ──


class TestProductiveLong:
    @pytest.fixture
    def scenario(self):
        return load_scenario("01_productive_long.yaml")

    @pytest.fixture
    def result(self, scenario):
        session = build_session(scenario)
        return process_session(session)

    def test_mode_is_full(self, result):
        mode, _ = result
        assert mode == Mode.FULL

    def test_candidate_count_in_range(self, result, scenario):
        _, candidates = result
        expected = scenario["expected"]["candidate_count"]
        assert expected["min"] <= len(candidates) <= expected["max"]

    def test_includes_telegram_skill_update(self, result):
        _, candidates = result
        telegram_updates = [
            c for c in candidates
            if c.type == CandidateType.SKILL_UPDATE and c.target_skill == "telegram"
        ]
        assert len(telegram_updates) >= 1
        assert any("Telethon" in c.description or "telethon" in c.description.lower()
                    for c in telegram_updates)

    def test_excludes_duplicate_memory(self, result):
        _, candidates = result
        memory_candidates = [c for c in candidates if c.type == CandidateType.FEEDBACK_MEMORY]
        for c in memory_candidates:
            assert "don't mock the database" not in c.proposed_content.lower()


# ── Scenario: short session ──


class TestShortSession:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("02_short_session.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_mode_is_fast(self, result):
        mode, _ = result
        assert mode == Mode.FAST

    def test_no_candidates_in_fast_mode(self, result):
        _, candidates = result
        assert len(candidates) == 0


# ── Scenario: rough session ──


class TestRoughSession:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("03_rough_session.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_mode_is_full(self, result):
        mode, _ = result
        assert mode == Mode.FULL

    def test_has_candidates(self, result):
        _, candidates = result
        assert len(candidates) >= 1


# ── Scenario: no skills used ──


class TestNoSkillsUsed:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("04_no_skills.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_mode_is_full(self, result):
        mode, _ = result
        assert mode == Mode.FULL

    def test_no_skill_update_candidates(self, result):
        _, candidates = result
        skill_updates = [c for c in candidates if c.type == CandidateType.SKILL_UPDATE]
        assert len(skill_updates) == 0

    def test_has_correction_candidate(self, result):
        _, candidates = result
        corrections = [c for c in candidates if c.source == SignalSource.USER_CORRECTION]
        assert len(corrections) >= 1
        assert any("single quotes" in c.proposed_content.lower() for c in corrections)


# ── Scenario: many learnings (cap test) ──


class TestManyLearnings:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("05_many_learnings.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_capped_at_max(self, result):
        _, candidates = result
        assert len(candidates) <= MAX_CANDIDATES

    def test_corrections_ranked_first(self, result):
        _, candidates = result
        if len(candidates) >= 2:
            first_sources = [c.source for c in candidates[:3]]
            assert SignalSource.USER_CORRECTION in first_sources

    def test_deduplicates_existing_telethon_memory(self, result):
        _, candidates = result
        for c in candidates:
            if c.type == CandidateType.FEEDBACK_MEMORY:
                assert "telethon directly for dms" not in c.proposed_content.lower()


# ── Scenario: duplicate memory ──


class TestDuplicateMemory:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("06_duplicate_memory.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_dedup_filters_known_insights(self, result):
        _, candidates = result
        memory_candidates = [c for c in candidates if c.type == CandidateType.FEEDBACK_MEMORY]
        for c in memory_candidates:
            assert "telethon directly" not in c.proposed_content.lower()


# ── Scenario: skill file conflict ──


class TestSkillConflict:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("07_skill_conflict.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_no_duplicate_skill_update(self, result):
        _, candidates = result
        pdf_updates = [
            c for c in candidates
            if c.type == CandidateType.SKILL_UPDATE and c.target_skill == "pdf-generation"
        ]
        assert len(pdf_updates) == 0


# ── Scenario: linear not configured ──


class TestLinearNotConfigured:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("08_linear_not_configured.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_no_linear_candidates(self, result):
        _, candidates = result
        linear = [c for c in candidates if c.type == CandidateType.LINEAR_TASK]
        assert len(linear) == 0

    def test_still_has_skill_update(self, result):
        _, candidates = result
        skill_updates = [c for c in candidates if c.type == CandidateType.SKILL_UPDATE]
        assert len(skill_updates) >= 1


# ── Scenario: cross-skill workflow ──


class TestCrossSkillWorkflow:
    @pytest.fixture
    def result(self):
        scenario = load_scenario("09_cross_skill_workflow.yaml")
        session = build_session(scenario)
        return process_session(session)

    def test_detects_workflow_pattern(self, result):
        _, candidates = result
        workflow = [c for c in candidates if c.source == SignalSource.WORKFLOW_PATTERN]
        assert len(workflow) >= 1


# ── Unit tests for individual functions ──


class TestDedup:
    def test_filters_exact_match(self):
        candidates = [
            _make_candidate("use Telethon directly for DMs"),
        ]
        memories = [Memory(name="x", content="Use Telethon directly for private DMs")]
        result = dedup_against_memory(candidates, memories)
        assert len(result) == 0

    def test_keeps_novel_content(self):
        candidates = [
            _make_candidate("always use single quotes"),
        ]
        memories = [Memory(name="x", content="Use Telethon for DMs")]
        result = dedup_against_memory(candidates, memories)
        assert len(result) == 1


class TestRanking:
    def test_corrections_before_failures(self):
        candidates = [
            _make_candidate("error", source=SignalSource.FAILED_SKILL),
            _make_candidate("correction", source=SignalSource.USER_CORRECTION),
        ]
        ranked = rank_candidates(candidates)
        assert ranked[0].source == SignalSource.USER_CORRECTION

    def test_failures_before_patterns(self):
        candidates = [
            _make_candidate("pattern", source=SignalSource.REPEATED_PATTERN),
            _make_candidate("failure", source=SignalSource.FAILED_SKILL),
        ]
        ranked = rank_candidates(candidates)
        assert ranked[0].source == SignalSource.FAILED_SKILL


class TestSkillConflictCheck:
    def test_detects_overlap(self):
        assert check_skill_conflict(
            "babel-lang: russian in YAML frontmatter",
            "Russian text needs babel-lang: russian in YAML frontmatter",
        )

    def test_no_overlap(self):
        assert not check_skill_conflict(
            "xelatex required for CJK fonts",
            "Russian text needs babel-lang: russian in YAML frontmatter",
        )


class TestLinearFilter:
    def test_removes_linear_when_not_configured(self):
        from retro_engine import Candidate, CandidateType, SignalSource
        candidates = [
            Candidate(
                type=CandidateType.LINEAR_TASK,
                source=SignalSource.FAILED_SKILL,
                label="Create task",
                description="Fix telegram bug",
            ),
            Candidate(
                type=CandidateType.SKILL_UPDATE,
                source=SignalSource.FAILED_SKILL,
                label="Update skill",
                description="Fix telegram bug",
            ),
        ]
        result = filter_linear_candidates(candidates, linear_configured=False)
        assert len(result) == 1
        assert result[0].type == CandidateType.SKILL_UPDATE


# ── Helpers ──


def _make_candidate(
    content: str,
    source: SignalSource = SignalSource.USER_CORRECTION,
) -> "Candidate":
    from retro_engine import Candidate, CandidateType
    return Candidate(
        type=CandidateType.FEEDBACK_MEMORY,
        source=source,
        label=content[:40],
        description=content,
        proposed_content=content,
    )