Skill Forge

"""Copy a staged skill into `.claude/skills/<name>/` and update the registry.

Create mode Step 5 and improve mode Step 4 both end here. The staging
step exists because Claude's own Write tool prompts whenever it targets
a path inside `.claude/` that doesn't already qualify as a real skill
dir (meaning: contain a SKILL.md). A fresh `.claude/skills/<name>/` has
no SKILL.md yet, so the first Write into it prompts even under
bypassPermissions. Improve mode is less affected — the target already
has SKILL.md — but we run it through the same pipe so create and improve
share one code path instead of two that drift.

What this script does:

  1. Validate the staging dir exists and has a SKILL.md.
  2. Parse + schema-check the frontmatter.
  3. Copy `<project>/.skill-forge/staging/<name>/` → `.claude/skills/<name>/`
     via shutil (a subprocess syscall, so no Claude tool permission layer).
  4. Consume `pending_eval_score` from state.json so the registry entry
     reflects the session's evaluator verdict instead of the default 0/8.
  5. Upsert the registry entry and persist.
  6. Remove the staging dir so nothing stale lingers.
  7. Clear the active draft (hooks stop injecting once draft is empty).

Mode semantics:

  create  target must NOT exist — a duplicate name is a user error.
  update  target MUST exist — rmtree then copytree for an atomic replace.
          Merging in place is tempting (shutil.copytree(..., dirs_exist_ok=True))
          but it leaves orphaned files behind when the improve session
          removed something from staging; full replacement is the only
          safe semantics given arbitrary file moves mid-session.
"""

from __future__ import annotations

import argparse
import shutil
import sys
from datetime import date
from pathlib import Path

from shared import (
    PENDING_EVAL_SCORE_KEY,
    SKILLS_DIR,
    draft_file,
    load_registry,
    load_state,
    parse_frontmatter,
    save_registry,
    save_state,
    staging_dir,
    state_file,
    upsert_skill,
)
from quick_validate import validate_skill as structural_validate


# ── planning ──────────────────────────────────────────────────────────


def _resolve_target(name: str, project_dir: Path) -> Path:
    return project_dir / SKILLS_DIR / name


def _parse_frontmatter(skill_md: Path, content: str) -> dict:
    fm = parse_frontmatter(content)
    if fm is None:
        raise ValueError(f"{skill_md} has no YAML frontmatter")
    if "name" not in fm or "description" not in fm:
        missing = {"name", "description"} - set(fm)
        raise ValueError(
            f"{skill_md} frontmatter missing required field(s): "
            f"{', '.join(sorted(missing))}"
        )
    return fm


def _load_frontmatter(skill_md: Path) -> dict:
    return _parse_frontmatter(skill_md, skill_md.read_text())


def _validate_mode(mode: str, target: Path) -> None:
    if mode == "create" and target.exists():
        raise FileExistsError(
            f"target already exists: {target} (use --mode update to replace)"
        )
    if mode == "update" and not target.exists():
        raise FileNotFoundError(
            f"target not found: {target} (use --mode create for new skill)"
        )


# ── core action ───────────────────────────────────────────────────────


def _append_changelog(target: Path, version: str, one_liner: str) -> None:
    """Prepend a dated entry — newest-first so `head` shows latest."""
    today = date.today().isoformat()
    entry = f"## {today} — v{version}\n- {one_liner.strip()}\n\n"
    path = target / "CHANGELOG.md"
    path.write_text(entry + path.read_text() if path.is_file() else entry)


def finalize(
    name: str,
    mode: str,
    project_dir: Path | None = None,
    changelog: str | None = None,
    bump: str = "patch",
) -> Path:
    """Move staged skill into place and update the registry.

    Returns the target path. Raises on precondition failures (missing
    staging, bad frontmatter, mode mismatch with existing target) — the
    caller should surface the message to the user without retrying.

    changelog: optional one-line entry appended to `<target>/CHANGELOG.md`
        with today's ISO date and the new version header — moves date /
        version / format concerns out of the prompt layer.
    bump: which semver segment to increment on update (default 'patch').
    """
    if project_dir is None:
        project_dir = Path.cwd()
    if mode not in {"create", "update"}:
        raise ValueError(f"mode must be 'create' or 'update', got {mode!r}")

    source = staging_dir(project_dir) / name
    if not source.is_dir():
        raise FileNotFoundError(
            f"staging dir not found: {source} "
            "(run init_staging.py first)"
        )

    skill_md = source / "SKILL.md"
    if not skill_md.is_file():
        raise FileNotFoundError(f"{skill_md} missing — staging is incomplete")

    content = skill_md.read_text()
    fm = _parse_frontmatter(skill_md, content)
    if fm["name"] != name:
        raise ValueError(
            f"frontmatter name {fm['name']!r} does not match requested {name!r}"
        )

    # Non-blocking structural warnings — surfaced so the user can fix on
    # the next iteration without gating the write.
    structural_warnings = structural_validate(source, content=content)

    target = _resolve_target(name, project_dir)
    _validate_mode(mode, target)

    # copytree refuses if target exists, so wipe first in update mode.
    if mode == "update":
        shutil.rmtree(target)
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, target)

    # Pop (not get) so a stale score from an earlier unrelated run can't
    # leak into this one. Only persist when a key was actually consumed.
    state_path = state_file(project_dir)
    state = load_state(state_path)
    pending_score = state.pop(PENDING_EVAL_SCORE_KEY, None)
    if pending_score is not None:
        save_state(state, state_path)

    registry_path = project_dir / SKILLS_DIR / "skill_registry.json"
    registry = load_registry(registry_path)
    new_version = upsert_skill(
        registry, fm, scope="project", eval_score=pending_score, bump=bump,
    )
    save_registry(registry, registry_path)

    if changelog:
        _append_changelog(target, new_version, changelog)

    # Clear staging — nothing under .skill-forge/staging/<name>/ should
    # outlive a successful finalize; leaving it risks confusing the user
    # into editing it expecting the changes to propagate.
    shutil.rmtree(source)

    # Clear the active draft so hooks stop injecting stale context. An
    # empty file rather than unlink, because downstream checks that look
    # at draft_file().is_file() shouldn't flip just because finalize ran.
    draft = draft_file(project_dir)
    if draft.is_file():
        draft.write_text("")

    _report(target, mode, pending_score, structural_warnings)
    return target


def _report(
    target: Path,
    mode: str,
    score: int | None,
    warnings: list[str],
) -> None:
    verb = "Created" if mode == "create" else "Updated"
    score_str = f"{score}/8" if score is not None else "—"
    print(f"[skill-forge] {verb} {target} (score: {score_str})")
    if warnings:
        print("[skill-forge] Validation notes:")
        for w in warnings:
            print(f"  - {w}")


# ── CLI ───────────────────────────────────────────────────────────────


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("name")
    parser.add_argument(
        "--mode",
        choices=["create", "update"],
        required=True,
    )
    parser.add_argument(
        "--changelog",
        default=None,
        help="One-line CHANGELOG entry; script adds date and version header.",
    )
    parser.add_argument(
        "--bump",
        choices=["patch", "minor", "major"],
        default="patch",
        help="Which semver segment to increment on update (default: patch).",
    )
    parser.add_argument(
        "--project-dir",
        type=Path,
        default=None,
        help="Override project root (for tests).",
    )
    args = parser.parse_args(argv)

    try:
        finalize(
            args.name,
            args.mode,
            project_dir=args.project_dir,
            changelog=args.changelog,
            bump=args.bump,
        )
    except (FileNotFoundError, FileExistsError, ValueError) as e:
        print(f"[skill-forge] finalize failed: {e}", file=sys.stderr)
        return 1

    return 0


if __name__ == "__main__":  # pragma: no cover — entry guard
    sys.exit(main())

"""Unified hook entrypoint for the three draft-injection hooks.

Replaces the previous three inline bash snippets in SKILL.md's frontmatter.
One Python script beats three shell snippets for a few reasons:

- Cross-platform. The bash versions used `${VAR:-default}`, `[ -f ]`, and
  `head -N` — all absent on Windows cmd.exe. Python works on every OS
  where skill-forge already requires python3.
- Single source of truth for workspace paths. Previously the shell hook
  and Python helpers each computed the project slug independently, which
  drifted on macOS (`/tmp` vs `/private/tmp`) and silently broke draft
  injection. Here we just import `shared.draft_file` / `insights_file`.
- Room to grow. Filtering (only inject when draft is non-empty, only on
  specific tool matches, etc.) stays readable in Python; bash conditionals
  get unreadable fast.

Modes:
  prompt   — UserPromptSubmit: print draft head + insights pointer
  pretool  — PreToolUse: print draft head (small, for attention anchoring)
  posttool — PostToolUse: nudge user to update draft after Write/Edit
"""

from __future__ import annotations

import argparse
import os
import sys
from pathlib import Path

# scripts/ is already on sys.path when invoked as a hook command
from shared import draft_file, insights_file


def _project_dir() -> Path:
    """Hook execution dir. Claude Code sets CLAUDE_PROJECT_DIR; fall back to cwd."""
    env = os.environ.get("CLAUDE_PROJECT_DIR")
    return Path(env) if env else Path.cwd()


def _read_nonempty(path: Path) -> str | None:
    """Return file contents only if non-empty; else None. Missing is None."""
    if not path.is_file():
        return None
    text = path.read_text()
    return text if text.strip() else None


def _head(text: str, lines: int) -> str:
    return "\n".join(text.splitlines()[:lines])


def inject_prompt(project_dir: Path, lines: int) -> str:
    """UserPromptSubmit body — full context reminder when a draft is active.

    Injected on every user prompt so Claude reorients after context shifts.
    Quiet when no draft exists (no noise between skill-forge sessions).
    """
    draft_path = draft_file(project_dir)
    text = _read_nonempty(draft_path)
    if text is None:
        return ""
    insights_path = insights_file(project_dir)
    return (
        "[skill-forge] ACTIVE SKILL DRAFT — current state:\n"
        f"{_head(text, lines)}\n\n"
        f"[skill-forge] Review {insights_path} for codebase context. "
        "Continue from current phase."
    )


def inject_pretool(project_dir: Path, lines: int) -> str:
    """PreToolUse body — short reminder before Read/Glob/Grep/Bash.

    Kept tiny (5 lines default) because this fires on every tool call and
    a 20-line dump floods the transcript. Header + Phase + Status is
    enough to anchor attention without blowing context.
    """
    draft_path = draft_file(project_dir)
    text = _read_nonempty(draft_path)
    if text is None:
        return ""
    return _head(text, lines)


def inject_posttool(project_dir: Path) -> str:
    """PostToolUse body — nudge to sync findings into draft after Write/Edit.

    Stat-only check (no read) — this fires on every Write/Edit, and the
    draft body is discarded; reading it just to test non-empty would
    scale with draft length per tool call.
    """
    draft_path = draft_file(project_dir)
    try:
        if draft_path.stat().st_size == 0:
            return ""
    except OSError:
        return ""
    return (
        f"[skill-forge] Update {draft_path} with what you just found. "
        "If a codebase pattern is confirmed, move it from insights.md into the draft steps."
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="skill-forge draft hook injector")
    parser.add_argument(
        "--mode",
        choices=["prompt", "pretool", "posttool"],
        required=True,
    )
    parser.add_argument(
        "--lines",
        type=int,
        default=None,
        help="Draft head line count. Defaults: prompt=40, pretool=5.",
    )
    args = parser.parse_args(argv)

    project_dir = _project_dir()

    if args.mode == "prompt":
        output = inject_prompt(project_dir, args.lines if args.lines is not None else 40)
    elif args.mode == "pretool":
        output = inject_pretool(project_dir, args.lines if args.lines is not None else 5)
    else:
        output = inject_posttool(project_dir)

    if output:
        print(output)
    return 0


if __name__ == "__main__":  # pragma: no cover — entry guard
    sys.exit(main())

"""Create mode draft initializer.

Generate the active draft file (under `<project>/.skill-forge/`) from name
and goal, serving as the attention anchor file for hooks.
"""

from __future__ import annotations

import sys
from pathlib import Path

# shared module from same directory
from shared import draft_file

DRAFT_TEMPLATE = """\
# {name} — IN PROGRESS
## Goal
{goal}
## Phase
Phase 1: codebase research
## Status
pending
"""


def create_draft(
    name: str,
    goal: str,
    project_dir: Path | None = None,
) -> None:
    """Create the active draft file.

    Auto-creates the workspace directory. Overwrites existing draft.
    """
    if project_dir is None:
        project_dir = Path.cwd()

    draft_path = draft_file(project_dir)
    draft_path.parent.mkdir(parents=True, exist_ok=True)
    draft_path.write_text(DRAFT_TEMPLATE.format(name=name, goal=goal))


def main(
    name: str | None = None,
    goal: str | None = None,
    project_dir: Path | None = None,
) -> None:
    """Entry point. Get name/goal from params or sys.argv, create draft."""
    if name is None:
        if len(sys.argv) < 3:
            print("Usage: init_draft.py <name> <goal>")
            sys.exit(1)
        name = sys.argv[1]
        goal = " ".join(sys.argv[2:])

    if goal is None:
        goal = ""

    create_draft(name, goal, project_dir=project_dir)
    print(f"[skill-forge] Draft initialized: {name}")


if __name__ == "__main__":  # pragma: no cover — entry guard
    main()

"""Improve mode session initializer.

Two jobs in one script:

  1. Seed staging from the live skill dir — `.claude/skills/<name>/*` →
     `.skill-forge/staging/<name>/*`. Claude's Edit calls during improve
     then land in staging, keeping `.claude/` untouched until
     `finalize_skill.py --mode update` copies the result back atomically.
  2. Copy the SKILL.md into the active draft workspace file as the
     attention anchor (the hooks re-read draft.md before every tool
     call, keeping Claude oriented mid-session).

Unified staging across create and improve means one finalize path for
both, and no branch where Claude edits a file the user didn't approve.
"""

from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path

# same directory — init_staging + shared both sit in scripts/
from init_staging import prepare as init_staging
from shared import SKILLS_DIR, draft_file


def init_improve_session(
    name: str,
    project_dir: Path | None = None,
) -> bool:
    """Initialize improve session.

    Seeds staging from the live skill dir and writes the draft. Returns
    False (no changes) when the target skill doesn't exist — the caller
    should surface that to the user since there's nothing to improve.
    """
    if project_dir is None:
        project_dir = Path.cwd()

    skill_dir = project_dir / SKILLS_DIR / name
    skill_file = skill_dir / "SKILL.md"
    if not skill_file.is_file():
        return False

    # Stage first so Edit/Write on the draft never races an unseeded
    # staging dir. init_staging wipes any stale staging for this name.
    init_staging(name, source=skill_dir, project_dir=project_dir)

    content = skill_file.read_text()
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    content += f"\n## Improve session — {timestamp}\n"

    draft_path = draft_file(project_dir)
    draft_path.parent.mkdir(parents=True, exist_ok=True)
    draft_path.write_text(content)
    return True


def main(
    name: str | None = None,
    project_dir: Path | None = None,
) -> None:
    """Entry point. Get name from params or sys.argv, initialize improve session."""
    if name is None:
        if len(sys.argv) < 2:
            print("Usage: init_improve.py <name>")
            sys.exit(1)
        name = sys.argv[1]

    success = init_improve_session(name, project_dir=project_dir)
    if success:
        print(f"[skill-forge] Improve session initialized: {name}")
    else:
        print(f"[skill-forge] Skill not found: {name}")


if __name__ == "__main__":  # pragma: no cover — entry guard
    main()

"""Staging directory initializer.

Creates `<project>/.skill-forge/staging/<name>/` so Claude can assemble a
skill there instead of writing directly into `.claude/skills/<name>/`.
Direct writes into a fresh `.claude/skills/<name>/` trigger permission
prompts even under bypassPermissions — the dir doesn't yet contain a
SKILL.md, so Claude Code's trust-boundary exemption doesn't cover it.

Two usage shapes:

  init_staging.py <name>
      Create mode: pre-writes a SKILL.md skeleton with frontmatter and
      empty section headers, so Claude iterates with Edit instead of
      authoring layout + frontmatter structure from scratch each time.

  init_staging.py <name> --source .claude/skills/<name>
      Improve mode: copy the existing skill dir into staging as the
      starting point for edits. Claude then Edit()s files in staging,
      and finalize_skill.py --mode update copies the result back.

Both shapes are safe to re-run — they blow away any existing contents at
the staging path first so a half-finished previous attempt can't leak in.
"""

from __future__ import annotations

import argparse
import re
import shutil
import sys
from pathlib import Path

from shared import staging_dir


_SKELETON = """\
---
name: {name}
description: >
  Use when <specific multi-step scenario — name real artifacts>.
  Even if the user just says <short phrase>, use when they mention <real phrasing>.
  Do NOT use when <simple single-step or adjacent task>.
user-invocable: true
---

# {name}

<one-paragraph intent: what this skill does and why it exists>

## Prerequisites
-

## Steps
1.
2.

## Verification
-

## Notes
-
"""


# Stricter than quick_validate's NAME_PATTERN (which accepts "123"): staging
# additionally requires a leading letter so directory names match what
# developers actually type. A digit-only name passes the Claude Code schema
# but is almost certainly a typo.
_NAME_RE = re.compile(r"^[a-z][a-z0-9]*(-[a-z0-9]+)*$")


def validate_name(name: str) -> None:
    """Raise ValueError if name isn't kebab-case. Matches frontmatter schema."""
    if not _NAME_RE.match(name):
        raise ValueError(
            f"skill name {name!r} must be lowercase kebab-case "
            "(e.g. 'generate-endpoint')"
        )


def prepare(
    name: str,
    source: Path | None = None,
    project_dir: Path | None = None,
) -> Path:
    """Create empty staging dir, or seed from `source` when provided.

    Idempotent: any existing content at the staging path is removed first.
    This matters for improve mode — a re-invocation after an aborted run
    must not merge old and new files under the same name.
    """
    validate_name(name)
    target = staging_dir(project_dir) / name
    if target.exists():
        shutil.rmtree(target)

    if source is not None:
        if not source.is_dir():
            raise FileNotFoundError(f"source skill dir not found: {source}")
        shutil.copytree(source, target)
    else:
        target.mkdir(parents=True, exist_ok=True)
        (target / "SKILL.md").write_text(_SKELETON.format(name=name))
    return target


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Initialize a skill staging dir.")
    parser.add_argument("name")
    parser.add_argument(
        "--source",
        type=Path,
        default=None,
        help="Existing skill dir to seed staging from (improve mode).",
    )
    parser.add_argument(
        "--project-dir",
        type=Path,
        default=None,
        help="Override project root (for tests).",
    )
    args = parser.parse_args(argv)

    try:
        target = prepare(args.name, source=args.source, project_dir=args.project_dir)
    except (ValueError, FileNotFoundError) as e:
        print(f"[skill-forge] init_staging failed: {e}", file=sys.stderr)
        return 1

    print(f"[skill-forge] Staging ready at {target}")
    return 0


if __name__ == "__main__":  # pragma: no cover — entry guard
    sys.exit(main())

"""Eval-driven description optimizer.

Iteratively evaluates a skill's trigger accuracy by *actually* running
`claude -p <query>` subprocesses and checking whether the Skill tool
fires with the target name (Anthropic skill-creator's approach).
Train/test split prevents overfitting; best iteration chosen by test score.

Key mechanics:
- Real execution eval (not LLM-as-judge): each query spawns a claude child
  process and we parse its stream-json output. Early-exits the instant the
  Skill invocation is detected — no need to wait for the full turn.
- Concurrent evaluation via ThreadPoolExecutor. Launch cadence is gated by
  a shared RateLimiter (shared.RateLimiter) so the RPM budget is
  respected across all threads.
- Per-round state persistence (opt_state.json) captures round history,
  FP/FN counts, convergence flag for external inspection / resume.
- Improve-prompt includes *history of past tries* so the generator is
  steered away from repeating the same variants.

Legacy LLM-judge constants (EVALUATE_TEMPLATE / IMPROVE_FN_GUIDANCE /
IMPROVE_FP_GUIDANCE / call_claude / _call_claude_once) are retained purely
for self_evolve.py's meta-optimizer, which scores prompt templates by
LLM-judge. The primary optimizer path does NOT invoke them.
"""

from __future__ import annotations

import argparse
import json
import random
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import asdict, dataclass
from pathlib import Path

# shared module from same directory
from shared import (
    DEFAULT_RPM,
    RateLimiter,
    log_stderr as _log,
    parse_frontmatter,
    run_subprocess,
)
from run_eval import find_project_root, run_single_query

# model used for improve/judge calls — change to test different models
CLAUDE_MODEL = "claude-sonnet-4-6"

# Defaults mirror Anthropic skill-creator's run_eval.py: 10 concurrent workers
# saturates a Tier-1 API without hitting 429s when combined with the RPM
# limiter. 3 runs per query averages out the per-run stochasticity of real
# eval (same query can trigger on run 1 and not on run 2). 30s covers a
# typical claude --print turn including stream events; early-exit makes
# most runs finish in ~3-5s.
DEFAULT_NUM_WORKERS = 10
DEFAULT_RUNS_PER_QUERY = 3
DEFAULT_TIMEOUT = 30
DEFAULT_TRIGGER_THRESHOLD = 0.5

# ── prompt templates (legacy: self_evolve.py meta-optimizer only) ──
# self_evolve imports these to optimize the templates themselves against a
# separate LLM-judge metric. The primary real-eval path does NOT use them.
# Placeholders: {description}, {query} are filled at call site.

EVALUATE_TEMPLATE = (
    "You are a skill activation filter for Claude Code. Your job: decide if a user query "
    "warrants invoking a specialized multi-step skill.\n\nSkill "
    "description:\n{description}\n\nUser query:\n{query}\n\nAsk yourself:\n1. Does the "
    "query require **coordinated, sequential actions** — not just a single operation?\n2. "
    "Would a developer expect this to take **multiple distinct phases** to complete?\n3. "
    "Does the skill description **closely match the problem being solved**, not just "
    "share keywords?\n\nIf all three are YES, output YES. Otherwise output NO.\n\nBias "
    "toward NO — a falsely triggered skill is more disruptive than a missed one."
)

IMPROVE_FN_GUIDANCE = (
    "Fix: the description failed to match these queries. Analyze what they have in common "
    "and add trigger patterns that capture that shared intent. Strategies:\n"
    "- Add 'Even if the user just says X, use this skill when they likely need Y' for "
    "cases where users understate their needs.\n"
    "- Name specific artifacts or actions from the missed queries (file types, commands, "
    "tools) so the description covers real phrasing, not just abstract concepts.\n"
    "- Front-load the most distinctive trigger keywords — Claude may truncate from the end."
)

IMPROVE_FP_GUIDANCE = (
    "Input: skill description that generated false positives.\nOutput: revised "
    "description with narrowed trigger scope.\n\nTransform rules — apply all that "
    "match:\n1. APPEND clause: \"Do NOT use when: <enumerate FP task types "
    "verbatim>\".\n2. REPLACE each vague verb (manage | handle | work with) with a "
    "multi-step scenario that requires the full workflow to be present.\n3. IF a keyword "
    "appears in both the FP queries and the description: qualify it with a condition "
    "(e.g., \"deploy\" → \"deploy with rollback and health checks\")."
)


# ── data loading ─────────────────────────────────────


def load_skill(skill_path: Path) -> tuple[str, str]:
    """Read SKILL.md, parse name and description from YAML frontmatter.

    skill_path can be file or directory (directory auto-resolves to SKILL.md).
    Failure or missing returns ("", "").
    """
    if skill_path.is_dir():
        skill_path = skill_path / "SKILL.md"

    if not skill_path.is_file():
        return ("", "")

    try:
        content = skill_path.read_text()
    except OSError:  # pragma: no cover — requires OS-level permission denial after is_file() passes
        return ("", "")

    fm = parse_frontmatter(content)
    if fm is None:
        return ("", "")

    return (fm.get("name", ""), fm.get("description", ""))


def load_evals(eval_path: Path) -> list[dict]:
    """Read trigger eval data in JSON array format.

    Failure returns empty list.
    """
    if not eval_path.is_file():
        return []

    try:
        data = json.loads(eval_path.read_text())
    except (json.JSONDecodeError, OSError):
        return []

    if not isinstance(data, list):
        return []

    return data


def split_train_test(
    evals: list[dict],
    ratio: float = 0.6,
    seed: int = 42,
) -> tuple[list[dict], list[dict]]:
    """Shuffle with fixed seed then split by ratio into train/test.

    20 items ratio=0.6 -> 12 train + 8 test.
    Empty list returns ([], []).
    """
    if not evals:
        return ([], [])

    shuffled = list(evals)
    random.Random(seed).shuffle(shuffled)

    split_index = int(len(shuffled) * ratio)
    return (shuffled[:split_index], shuffled[split_index:])


# ── Claude CLI wrapper ───────────────────────────────


def _call_claude_once(prompt: str) -> str:
    """Single claude --print invocation; empty on failure.

    cwd=/tmp keeps the CLI from auto-loading project CLAUDE.md and leaking
    house style into the response.
    """
    cmd = ["claude", "--model", CLAUDE_MODEL, "--print", "-p", prompt]
    return run_subprocess(cmd, timeout=60, cwd="/tmp")


def call_claude(prompt: str) -> str:
    """Call claude --print with a 2s-backoff retry on empty response."""
    result = _call_claude_once(prompt)
    if result:
        return result
    time.sleep(2)
    return _call_claude_once(prompt)


# ── DSPy-inspired data structures ───────────────────


@dataclass
class RoundRecord:
    """Per-round optimization metrics for state persistence."""

    round: int
    description: str
    train_score: float
    test_score: float
    false_positive_count: int
    false_negative_count: int


@dataclass
class OptState:
    """Persistent optimization state across runs."""

    skill_name: str
    best_score: float
    best_description: str
    current_round: int
    converged: bool
    rounds: list[RoundRecord]


def load_opt_state(path: Path) -> OptState | None:
    """Load opt state from JSON. Missing/corrupt/schema-mismatch returns None.

    Not called from main() — provided for external callers (SKILL.md improve mode
    inspect/resume, CLI status command). Optimizer always starts fresh per run.
    """
    try:
        data = json.loads(path.read_text())
    except (json.JSONDecodeError, OSError):
        return None

    if not isinstance(data, dict) or "skill_name" not in data:
        return None

    try:
        rounds = [RoundRecord(**r) for r in data.get("rounds", [])]
        return OptState(
            skill_name=data["skill_name"],
            best_score=data["best_score"],
            best_description=data["best_description"],
            current_round=data["current_round"],
            converged=data["converged"],
            rounds=rounds,
        )
    except (KeyError, TypeError):
        return None


def save_opt_state(state: OptState, path: Path) -> None:
    """Persist opt state. Auto-creates parent directory."""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(asdict(state), indent=2))
    _log(f"  State saved: round={state.current_round} best={state.best_score:.2f} path={path}")


def classify_failures(failures: list[dict]) -> tuple[list[dict], list[dict]]:
    """Split failures into (false_positives, false_negatives).

    FP: should_trigger=False but got=True (description too broad).
    FN: should_trigger=True but got=False (description misses scenario).
    """
    false_positives = [f for f in failures if f["should_trigger"] is False]
    false_negatives = [f for f in failures if f["should_trigger"] is True]
    return (false_positives, false_negatives)


# ── aggregated evaluation ────────────────────────────


def evaluate_single(
    description: str,
    query: str,
    runs: int = DEFAULT_RUNS_PER_QUERY,
    *,
    skill_name: str = "",
    project_root: Path | None = None,
    timeout: int = DEFAULT_TIMEOUT,
    threshold: float = DEFAULT_TRIGGER_THRESHOLD,
    limiter: RateLimiter | None = None,
    model: str | None = None,
) -> bool:
    """Trigger decision for a single query via N real-execution runs.

    Runs `claude -p <query>` `runs` times and computes trigger_rate. Returns
    True iff trigger_rate >= threshold. Stream early-exit keeps most runs
    cheap; the rate limiter gates cross-thread launch cadence.
    """
    if runs <= 0:
        return False

    triggers = 0
    for _ in range(runs):
        if limiter is not None:
            limiter.throttle()
        if run_single_query(
            query, skill_name, description,
            timeout=timeout, project_root=project_root, model=model,
        ):
            triggers += 1

    return (triggers / runs) >= threshold


def evaluate_set(
    description: str,
    eval_set: list[dict],
    *,
    skill_name: str = "",
    project_root: Path | None = None,
    num_workers: int = DEFAULT_NUM_WORKERS,
    runs: int = DEFAULT_RUNS_PER_QUERY,
    timeout: int = DEFAULT_TIMEOUT,
    threshold: float = DEFAULT_TRIGGER_THRESHOLD,
    limiter: RateLimiter | None = None,
    model: str | None = None,
) -> tuple[float, list[dict]]:
    """Score against entire eval set concurrently.

    ThreadPoolExecutor runs evaluate_single across queries; the shared
    limiter ensures total launch RPM stays within budget. Workers default
    to min(len(eval_set), num_workers).

    Returns (accuracy 0.0-1.0, failure list). Each failure:
    {query, should_trigger, got}.
    """
    if not eval_set:
        return (1.0, [])

    workers = max(1, min(num_workers, len(eval_set)))

    def _score_one(item: dict) -> tuple[dict, bool]:
        triggered = evaluate_single(
            description, item["query"], runs,
            skill_name=skill_name,
            project_root=project_root,
            timeout=timeout,
            threshold=threshold,
            limiter=limiter,
            model=model,
        )
        return (item, triggered)

    failures: list[dict] = []
    correct = 0
    with ThreadPoolExecutor(max_workers=workers) as executor:
        for future in as_completed([executor.submit(_score_one, item) for item in eval_set]):
            try:
                item, got = future.result()
            except Exception as exc:  # noqa: BLE001 — best-effort, log and continue
                _log(f"  Query failed: {exc}")
                continue
            if got == item["should_trigger"]:
                correct += 1
            else:
                failures.append({
                    "query": item["query"],
                    "should_trigger": item["should_trigger"],
                    "got": got,
                })

    accuracy = correct / len(eval_set)
    return (accuracy, failures)


def improve_description(
    description: str,
    failures: list[dict],
    *,
    classified: tuple[list[dict], list[dict]] | None = None,
    prior_attempts: list[str] | None = None,
) -> str:
    """Improve description via claude based on failure cases.

    No failures -> return original description (no claude call).
    Empty result or >300 chars -> fallback to original.
    classified: pre-computed (false_positives, false_negatives) to avoid re-classification.
    prior_attempts: previously-tried descriptions. Passed to the LLM as an
    explicit "don't repeat these" list — skill-creator's optimizer shows
    convergence is much faster when each round explores a structurally
    different axis rather than paraphrasing the last try.
    """
    if not failures:
        return description

    false_positives, false_negatives = classified or classify_failures(failures)

    prompt_parts = [
        "Improve this skill description for better trigger accuracy.",
        f"\nCurrent description:\n{description}",
    ]

    # FN direction: description misses scenarios → add concrete trigger patterns
    if false_negatives:
        queries = "\n".join(f"- {f['query']}" for f in false_negatives)
        prompt_parts.append(
            f"\nFalse negatives — missed these scenarios (should trigger but did not):\n{queries}\n"
            f"{IMPROVE_FN_GUIDANCE}"
        )

    # FP direction: description too broad → add DO NOT use clauses
    if false_positives:
        queries = "\n".join(f"- {f['query']}" for f in false_positives)
        prompt_parts.append(
            f"\nFalse positives — triggered when it should not:\n{queries}\n"
            f"{IMPROVE_FP_GUIDANCE}"
        )

    if prior_attempts:
        prior = "\n".join(f"- {p}" for p in prior_attempts)
        prompt_parts.append(
            f"\nPrevious attempts (do NOT repeat or paraphrase these; "
            f"try a structurally different axis):\n{prior}"
        )

    prompt_parts.append(
        "\nWrite an improved description under 250 characters. "
        "Output ONLY the new description text, nothing else."
    )

    result = call_claude("\n".join(prompt_parts))

    # prompt asks for 250 chars, but allow 300 tolerance — LLM may slightly exceed, soft fallback beats hard truncation
    if not result or len(result) > 300:
        return description

    return result


# ── optimization loop ────────────────────────────────


def run_optimization(
    description: str,
    train_set: list[dict],
    test_set: list[dict],
    max_iterations: int = 5,
    state_path: Path | None = None,
    skill_name: str = "",
    *,
    project_root: Path | None = None,
    num_workers: int = DEFAULT_NUM_WORKERS,
    runs: int = DEFAULT_RUNS_PER_QUERY,
    timeout: int = DEFAULT_TIMEOUT,
    threshold: float = DEFAULT_TRIGGER_THRESHOLD,
    limiter: RateLimiter | None = None,
    model: str | None = None,
) -> dict:
    """Iteratively optimize description, select best by test score.

    Each round: evaluate train -> evaluate test -> record best -> persist state.
    Perfect train (1.0) -> early stop.
    state_path: when provided, saves OptState after each round for history/convergence tracking.
    Returns {best_description, best_test_score, iterations, rounds, converged}.
    """
    best_description = description
    best_test_score = -1.0
    current_description = description
    rounds: list[RoundRecord] = []
    converged = False
    prior_attempts: list[str] = [description]  # feeds improve_description's anti-repeat clause

    common = dict(
        skill_name=skill_name,
        project_root=project_root,
        num_workers=num_workers,
        runs=runs,
        timeout=timeout,
        threshold=threshold,
        limiter=limiter,
        model=model,
    )

    iteration = 0
    for iteration in range(1, max_iterations + 1):
        _log(f"Iteration {iteration}/{max_iterations}")

        train_score, train_failures = evaluate_set(current_description, train_set, **common)
        _log(f"  Train score: {train_score:.2f}")

        test_score, _ = evaluate_set(current_description, test_set, **common)
        _log(f"  Test score:  {test_score:.2f}")

        # DSPy-inspired: track FP/FN counts per round for structured history
        false_positives, false_negatives = classify_failures(train_failures)
        round_record = RoundRecord(
            round=iteration,
            description=current_description,
            train_score=train_score,
            test_score=test_score,
            false_positive_count=len(false_positives),
            false_negative_count=len(false_negatives),
        )
        rounds.append(round_record)

        _log(f"  FP: {round_record.false_positive_count}  FN: {round_record.false_negative_count}")

        if test_score > best_test_score:
            best_test_score = test_score
            best_description = current_description

        # convergence: computed unconditionally, used by both state persistence and early-stop
        converged = train_score >= 1.0

        # persist state after each round (partial runs still capture history)
        if state_path is not None:
            save_opt_state(
                OptState(
                    skill_name=skill_name,
                    best_score=best_test_score,
                    best_description=best_description,
                    current_round=iteration,
                    converged=converged,
                    rounds=rounds,
                ),
                state_path,
            )

        # perfect train -> stop (evaluate test before break to include final test score in best selection)
        if converged:
            _log("  Perfect train score, stopping early.")
            break

        new_description = improve_description(
            current_description, train_failures,
            classified=(false_positives, false_negatives),
            prior_attempts=prior_attempts,
        )
        if new_description != current_description:
            prior_attempts.append(new_description)
        current_description = new_description
        _log(f"  Improved: {current_description[:80]}...")

    return {
        "best_description": best_description,
        "best_test_score": best_test_score,
        "iterations": iteration,
        "rounds": [asdict(r) for r in rounds],
        "converged": converged,
    }


# ── entry point ──────────────────────────────────────


def main() -> None:
    """CLI entry point.

    Parse args, load data, run optimization, output JSON to stdout.
    Skill or evals load failure -> exit(1).
    """
    parser = argparse.ArgumentParser(
        description="Eval-driven skill description optimizer (real-execution eval)"
    )
    parser.add_argument("--skill-path", required=True, help="Path to SKILL.md or skill directory")
    parser.add_argument("--eval-set", required=True, help="Path to trigger_evals.json")
    parser.add_argument("--max-iterations", type=int, default=5, help="Max optimization iterations")
    parser.add_argument("--num-workers", type=int, default=DEFAULT_NUM_WORKERS,
                        help=f"Concurrent eval workers (default {DEFAULT_NUM_WORKERS})")
    parser.add_argument("--runs-per-query", type=int, default=DEFAULT_RUNS_PER_QUERY,
                        help=f"Runs per eval query for trigger-rate averaging (default {DEFAULT_RUNS_PER_QUERY})")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT,
                        help=f"Per-query timeout seconds (default {DEFAULT_TIMEOUT})")
    parser.add_argument("--trigger-threshold", type=float, default=DEFAULT_TRIGGER_THRESHOLD,
                        help=f"Trigger-rate pass threshold (default {DEFAULT_TRIGGER_THRESHOLD})")
    parser.add_argument("--rpm", type=int, default=DEFAULT_RPM,
                        help=f"API RPM cap (default {DEFAULT_RPM}; raise on higher Anthropic tiers)")
    parser.add_argument("--model", default=None, help="Model for claude -p (default: user config)")
    args = parser.parse_args()

    # load skill
    skill_path = Path(args.skill_path)
    name, original_description = load_skill(skill_path)
    if not name or not original_description:
        _log("ERROR: Failed to load skill or missing name/description")
        sys.exit(1)

    # derive state path: <skill_dir>/.opt/opt_state.json
    skill_dir = skill_path if skill_path.is_dir() else skill_path.parent
    state_path = skill_dir / ".opt" / "opt_state.json"

    # load evals
    evals = load_evals(Path(args.eval_set))
    if not evals:
        _log("ERROR: Failed to load eval set or empty")
        sys.exit(1)

    # 60/40 split
    train_set, test_set = split_train_test(evals, ratio=0.6, seed=42)
    _log(f"Loaded {len(evals)} evals: {len(train_set)} train, {len(test_set)} test")

    # run optimization with state persistence
    limiter = RateLimiter(rpm=args.rpm)
    project_root = find_project_root(skill_path if skill_path.is_dir() else skill_path.parent)
    result = run_optimization(
        original_description,
        train_set,
        test_set,
        max_iterations=args.max_iterations,
        state_path=state_path,
        skill_name=name,
        project_root=project_root,
        num_workers=args.num_workers,
        runs=args.runs_per_query,
        timeout=args.timeout,
        threshold=args.trigger_threshold,
        limiter=limiter,
        model=args.model,
    )

    # output result JSON
    output = {
        "skill_name": name,
        "original_description": original_description,
        "best_description": result["best_description"],
        "best_test_score": result["best_test_score"],
        "iterations": result["iterations"],
    }
    print(json.dumps(output, indent=2))


if __name__ == "__main__":  # pragma: no cover — entry guard
    main()

"""Phase 0 context loader.

Merge Phase 0 commands from SKILL.md into a single call:
1. report installed version + stale-cache warning
2. read active draft head
3. run session catchup
4. list registered skills
5. read registry summary

Output structured report with section headers to stdout.
"""

from __future__ import annotations

import json
import os
from pathlib import Path

# shared module from same directory
from shared import REGISTRY_FILE, SKILLS_DIR, draft_file, load_registry
from skill_catchup import main as catchup_main

# default max lines
DEFAULT_DRAFT_LINES = 20

EMBED_VERSION_FILE = Path(".claude/hooks/skill-forge/version.json")
PLUGIN_MANIFEST_REL = Path(".claude-plugin/plugin.json")
PLUGIN_CACHE_DIR = Path.home() / ".claude" / "plugins" / "cache" / "skill-forge" / "skill-forge"


def load_draft_head(project_dir: Path, max_lines: int = DEFAULT_DRAFT_LINES) -> str:
    """Read first N lines of the active draft workspace file.

    File not found returns empty string.
    """
    draft = draft_file(project_dir)
    if not draft.is_file():
        return ""
    lines = draft.read_text().splitlines()[:max_lines]
    return "\n".join(lines)


def run_catchup(project_dir: Path) -> str:
    """Run session catchup scan directly (same process, no subprocess overhead).

    Returns report string (empty if nothing found).
    """
    return catchup_main(cwd=str(project_dir))


def load_skills_list(project_dir: Path) -> str:
    """List subdirectory names under .claude/skills/.

    SKILL.md presence is the skill anchor — filters out per-skill
    `-workspace/` helpers and stray dirs without a manifest.
    Directory not found returns empty string.
    """
    skills_dir = project_dir / SKILLS_DIR
    if not skills_dir.is_dir():
        return ""
    names = sorted(p.parent.name for p in skills_dir.glob("*/SKILL.md"))
    if not names:
        return ""
    return "\n".join(names)


def _read_json(path: Path) -> dict:
    """Read and parse JSON file. Missing or malformed returns empty dict."""
    try:
        return json.loads(path.read_text())
    except (FileNotFoundError, json.JSONDecodeError, ValueError, OSError):
        return {}


def detect_install(
    project_dir: Path,
    plugin_root: str | None = None,
    cache_dir: Path | None = None,
) -> str:
    """Identify install mode + version + stale cache warning.

    Order: plugin env (CLAUDE_PLUGIN_ROOT set) -> embed version.json -> dev
    repo manifest. Stale-cache warning covers the case where Claude's skill
    lookup can fall back to an older cached plugin version instead of the
    one registered in SKILL.md.
    """
    if plugin_root is None:
        plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT")
    if cache_dir is None:
        cache_dir = PLUGIN_CACHE_DIR

    cached_versions = (
        sorted(p.name for p in cache_dir.iterdir() if p.is_dir())
        if cache_dir.is_dir()
        else []
    )
    lines: list[str] = []

    if plugin_root:
        manifest = _read_json(Path(plugin_root) / PLUGIN_MANIFEST_REL)
        version = manifest.get("version", "unknown")
        lines.append(f"skill-forge {version} [plugin]")
        if len(cached_versions) > 1:
            lines.append(
                f"WARNING: multiple plugin cache versions present ({', '.join(cached_versions)}). "
                f"Run `/plugin update skill-forge` or `rm -rf {cache_dir}` then reinstall."
            )
        return "\n".join(lines)

    version_data = _read_json(project_dir / EMBED_VERSION_FILE)
    if version_data:
        version = version_data.get("version", "unknown")
        installed = version_data.get("installed", "?")
        lines.append(f"skill-forge {version} [embed, installed {installed}]")
        if cached_versions:
            lines.append(
                f"WARNING: marketplace plugin cache still present at {cache_dir} "
                f"(versions: {', '.join(cached_versions)}). Embed install is authoritative; "
                f"delete the cache to stop Claude's skill lookup from falling back to it."
            )
        return "\n".join(lines)

    for ancestor in [project_dir, *project_dir.parents]:
        manifest_path = ancestor / PLUGIN_MANIFEST_REL
        if manifest_path.is_file():
            manifest = _read_json(manifest_path)
            version = manifest.get("version", "unknown")
            lines.append(f"skill-forge {version} [dev @ {ancestor}]")
            break

    return "\n".join(lines)


def load_registry_summary(project_dir: Path) -> str:
    """Read skill_registry.json and format summary.

    File missing/corrupted returns empty string. Empty skills list returns hint text.
    """
    data = load_registry(project_dir / REGISTRY_FILE)
    skills = data.get("skills", [])
    if not skills:
        return "No skills registered."

    lines = []
    for skill in skills:
        name = skill.get("name", "?")
        version = skill.get("version", "?")
        updated = skill.get("updated", "?")
        lines.append(f"  {name}  v{version}  updated {updated}")
    return "\n".join(lines)


def main(
    project_dir: Path | None = None,
) -> None:
    """Entry point. Output structured report to stdout.

    project_dir: project root (defaults to cwd).
    """
    if project_dir is None:
        project_dir = Path.cwd()

    sections: list[str] = []

    # 1. install version — first so warnings show at the top
    version = detect_install(project_dir)
    if version:
        sections.append(f"=== Version ===\n{version}")

    # 2. active draft
    draft = load_draft_head(project_dir)
    if draft:
        sections.append(f"=== Draft ===\n{draft}")

    # 3. Session catchup
    catchup = run_catchup(project_dir)
    if catchup:
        sections.append(f"=== Catchup ===\n{catchup}")

    # 4. skills directory
    skills = load_skills_list(project_dir)
    if skills:
        sections.append(f"=== Skills ===\n{skills}")

    # 5. registry summary
    registry = load_registry_summary(project_dir)
    if registry:
        sections.append(f"=== Registry ===\n{registry}")

    if sections:
        print("\n\n".join(sections))
    else:
        print("skill-forge: no active draft, no skills registered.")


if __name__ == "__main__":  # pragma: no cover — entry guard
    main()

"""Structural YAML frontmatter validator for SKILL.md.

Ported from Anthropic's skill-creator quick_validate.py. Runs before registry
upsert in PostToolUse hook so field typos / schema violations never land
silently. Non-blocking — returns a list of error strings the hook surfaces as
systemMessage warnings. CLI entry exits 0 on empty list, 1 otherwise.

Enforced by Claude Code spec:
- name: kebab-case, ≤ 64 chars
- description: ≤ 1024 chars, no angle brackets (breaks YAML parsing downstream)
- only these top-level keys: name, description, license, allowed-tools,
  user-invocable, metadata, compatibility, hooks
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

# Claude Code SKILL.md spec — any other top-level key indicates a typo or
# non-standard extension we don't want silently accepted into the registry.
ALLOWED_KEYS = frozenset({
    "name",
    "description",
    "license",
    "allowed-tools",
    "user-invocable",
    "metadata",
    "compatibility",
    "hooks",
})

MAX_NAME_CHARS = 64
MAX_DESCRIPTION_CHARS = 1024
MAX_COMPATIBILITY_CHARS = 500
NAME_PATTERN = re.compile(r"^[a-z0-9-]+$")


def _extract_frontmatter(content: str) -> tuple[dict | None, str | None]:
    """Extract raw frontmatter dict plus a parse error on failure.

    Prefers PyYAML (dicts-of-dicts, quoted strings, list values all work). Falls
    back to shared.parse_frontmatter which handles flat key/value only — enough
    for the required-field checks even when PyYAML is absent.
    """
    if not content.startswith("---"):
        return None, "No YAML frontmatter found"

    match = re.match(r"^---\s*\n(.*?)\n---", content, re.DOTALL)
    if not match:
        return None, "Invalid frontmatter format"

    raw = match.group(1)
    try:
        import yaml  # type: ignore
        data = yaml.safe_load(raw)
    except ImportError:
        from shared import parse_frontmatter
        data = parse_frontmatter(content)
    except Exception as exc:  # noqa: BLE001 — yaml.YAMLError subclasses vary
        return None, f"Invalid YAML in frontmatter: {exc}"

    if not isinstance(data, dict):
        return None, "Frontmatter must be a YAML dictionary"

    return data, None


def validate_skill(skill_path: Path | str, content: str | None = None) -> list[str]:
    """Validate a skill directory's SKILL.md. Empty list = valid.

    Accepts either the directory or SKILL.md path directly so callers don't
    have to normalize. Pass `content` to skip the file read when the caller
    already has the bytes in hand (e.g. the PostToolUse hook).
    """
    path = Path(skill_path)
    skill_md = path / "SKILL.md" if path.is_dir() else path

    if content is None:
        if not skill_md.is_file():
            return ["SKILL.md not found"]
        try:
            content = skill_md.read_text()
        except OSError as exc:
            return [f"Cannot read SKILL.md: {exc}"]

    frontmatter, err = _extract_frontmatter(content)
    if err is not None:
        return [err]
    assert frontmatter is not None  # err-None invariant

    errors: list[str] = []

    unexpected = set(frontmatter.keys()) - ALLOWED_KEYS
    if unexpected:
        errors.append(
            f"Unexpected frontmatter key(s): {', '.join(sorted(unexpected))}. "
            f"Allowed: {', '.join(sorted(ALLOWED_KEYS))}"
        )

    if "name" not in frontmatter:
        errors.append("Missing required field: name")
    else:
        errors.extend(_validate_name(frontmatter["name"]))

    if "description" not in frontmatter:
        errors.append("Missing required field: description")
    else:
        errors.extend(_validate_description(frontmatter["description"]))

    if "compatibility" in frontmatter:
        errors.extend(_validate_compatibility(frontmatter["compatibility"]))

    return errors


def _validate_name(name: object) -> list[str]:
    """Kebab-case + length. Non-string type is a hard error."""
    if not isinstance(name, str):
        return [f"name must be a string, got {type(name).__name__}"]
    stripped = name.strip()
    if not stripped:
        return ["name cannot be empty"]
    if not NAME_PATTERN.match(stripped):
        return [f"name '{stripped}' must be kebab-case (lowercase letters, digits, hyphens only)"]
    if stripped.startswith("-") or stripped.endswith("-") or "--" in stripped:
        return [f"name '{stripped}' cannot start/end with hyphen or contain consecutive hyphens"]
    if len(stripped) > MAX_NAME_CHARS:
        return [f"name too long ({len(stripped)} chars, max {MAX_NAME_CHARS})"]
    return []


def _validate_description(desc: object) -> list[str]:
    """Length + angle-bracket ban (description is interpolated into prompts)."""
    if not isinstance(desc, str):
        return [f"description must be a string, got {type(desc).__name__}"]
    stripped = desc.strip()
    if not stripped:
        return ["description cannot be empty"]
    errors: list[str] = []
    if "<" in stripped or ">" in stripped:
        errors.append("description cannot contain angle brackets (< or >)")
    if len(stripped) > MAX_DESCRIPTION_CHARS:
        errors.append(f"description too long ({len(stripped)} chars, max {MAX_DESCRIPTION_CHARS})")
    return errors


def _validate_compatibility(compat: object) -> list[str]:
    """Optional free-form string capped at 500 chars."""
    if not isinstance(compat, str):
        return [f"compatibility must be a string, got {type(compat).__name__}"]
    if len(compat) > MAX_COMPATIBILITY_CHARS:
        return [f"compatibility too long ({len(compat)} chars, max {MAX_COMPATIBILITY_CHARS})"]
    return []


def main() -> None:  # pragma: no cover — CLI entry guard
    if len(sys.argv) != 2:
        print("Usage: python quick_validate.py <skill_directory_or_SKILL.md>", file=sys.stderr)
        sys.exit(2)

    errors = validate_skill(sys.argv[1])
    if not errors:
        print("Skill is valid!")
        sys.exit(0)
    for err in errors:
        print(f"ERROR: {err}", file=sys.stderr)
    sys.exit(1)


if __name__ == "__main__":  # pragma: no cover
    main()

"""Record session evaluator score → state.json.

The evaluator runs in-session (Claude scores draft against rubric in SKILL.md).
The score lives only in chat output; the PostToolUse hook that upserts the
registry on SKILL.md write has no access to it. This script bridges them:
Claude calls it before Write, the hook consumes `pending_eval_score` from
state.json on the next SKILL.md write and clears it.

Usage: python3 record_eval_score.py <score 0-8>
"""

from __future__ import annotations

import sys

from shared import PENDING_EVAL_SCORE_KEY, load_state, save_state

MAX_SCORE = 8


def record_score(score: int) -> None:
    """Persist pending eval score in workspace state.

    Hook reads the key on next SKILL.md write and clears it.
    Out-of-range raises ValueError to fail loudly rather than silently mis-record.
    """
    if not 0 <= score <= MAX_SCORE:
        raise ValueError(f"score must be 0..{MAX_SCORE}, got {score}")

    state = load_state()
    state[PENDING_EVAL_SCORE_KEY] = score
    save_state(state)


def main(argv: list[str] | None = None) -> None:
    if argv is None:
        argv = sys.argv[1:]
    if len(argv) != 1:
        print("Usage: record_eval_score.py <score 0-8>")
        sys.exit(1)
    try:
        score = int(argv[0])
    except ValueError:
        print(f"score must be an integer 0..{MAX_SCORE}, got {argv[0]!r}")
        sys.exit(1)

    try:
        record_score(score)
    except ValueError as e:
        print(str(e))
        sys.exit(1)
    print(f"[skill-forge] Recorded eval score: {score}/{MAX_SCORE}")


if __name__ == "__main__":  # pragma: no cover — entry guard
    main()

"""Rename a skill end-to-end.

One Python entry point so `/rename` never has to shell out `mv`, Write the
registry directly, or Edit files under `.claude/skills/<name>-workspace/` —
all three paths trigger permission prompts (mv has no stable allowlist; the
registry file and `<name>-workspace/` sit outside any real skill dir and the
`.claude/` exemption does not recurse there).

Runs as a subprocess so pathlib/shutil write/rename bypass Claude's tool
permission layer entirely. `--dry-run` prints the plan for confirmation
without touching disk.
"""

from __future__ import annotations

import argparse
import json
import shutil
import sys
from datetime import date
from pathlib import Path

# shared module from same directory
from shared import (
    REGISTRY_FILE,
    SKILLS_DIR,
    USER_SKILLS_DIR,
    draft_file,
    load_registry,
    save_registry,
)


# ── Scope resolution ──────────────────────────────────────────────


def resolve_skills_root(
    scope: str | None,
    project_dir: Path,
) -> tuple[Path, str]:
    """Return (skills_dir, scope_label).

    `scope` in {"project", "user", None}. None auto-detects: project dir first
    if it has `.claude/skills/`, else user dir.
    """
    project_skills = project_dir / SKILLS_DIR
    if scope == "project":
        return project_skills, "project"
    if scope == "user":
        return USER_SKILLS_DIR, "user"
    if project_skills.is_dir():
        return project_skills, "project"
    return USER_SKILLS_DIR, "user"


# ── Plan building ─────────────────────────────────────────────────


def _scan_dir(dir_path: Path, old_name: str) -> list[tuple[Path, str, int]]:
    """Return (path, text, count) for every file under dir_path that
    contains old_name. Binary/unreadable files are skipped. Caching text
    here lets execute_plan skip a second read per file.
    """
    hits: list[tuple[Path, str, int]] = []
    for path in sorted(dir_path.rglob("*")):
        if not path.is_file():
            continue
        try:
            text = path.read_text()
        except (OSError, UnicodeDecodeError):
            continue
        count = text.count(old_name)
        if count > 0:
            hits.append((path, text, count))
    return hits


def build_plan(
    old_name: str,
    new_name: str,
    skills_root: Path,
    project_dir: Path,
) -> dict:
    """Collect changes without touching disk.

    Validates preconditions, scans every file in the skill dir for old_name
    occurrences, and checks whether the active draft references old_name.

    Returns dict with keys: old_name, new_name, errors[], warnings[],
    file_edits[(path, text, count)], dir_renames[], registry fields.
    """
    errors: list[str] = []
    warnings: list[str] = []

    if old_name == new_name:
        errors.append("old-name and new-name are identical")

    old_dir = skills_root / old_name
    new_dir = skills_root / new_name
    if not old_dir.is_dir():
        errors.append(f"skill dir not found: {old_dir}")
    if new_dir.exists():
        errors.append(f"target already exists: {new_dir}")

    registry_path = skills_root / REGISTRY_FILE.name
    registry = load_registry(registry_path)
    entry = next(
        (s for s in registry.get("skills", []) if s.get("name") == old_name),
        None,
    )
    if entry is None:
        errors.append(f"registry has no entry named {old_name!r}")

    # Active draft guardrail — if mid-session on this skill, renaming mid-flight
    # corrupts the draft's implicit target.
    draft = draft_file(project_dir)
    if draft.is_file() and old_name in draft.read_text():
        errors.append(
            f"active draft {draft} still references {old_name!r}; "
            f"finish the create/improve session before renaming"
        )

    file_edits: list[tuple[Path, str, int]] = []
    dir_renames: list[tuple[Path, Path]] = []

    if old_dir.is_dir():
        file_edits.extend(_scan_dir(old_dir, old_name))
        dir_renames.append((old_dir, new_dir))

    return {
        "old_name": old_name,
        "new_name": new_name,
        "errors": errors,
        "warnings": warnings,
        "file_edits": file_edits,
        "dir_renames": dir_renames,
        "registry_path": registry_path,
        "registry": registry,
        "registry_entry": entry,
    }


# ── Execution ─────────────────────────────────────────────────────


def execute_plan(plan: dict) -> None:
    """Apply plan in safe order: content edits → dir renames → registry.

    Content first so paths are still valid. Registry last so a mid-execution
    crash leaves either the old state (if pre-rename) or a consistent on-disk
    layout that matches the registry (if post-rename, rerun fixes registry).
    """
    old_name = plan["old_name"]
    new_name = plan["new_name"]

    for path, text, _ in plan["file_edits"]:
        path.write_text(text.replace(old_name, new_name))

    for src, dst in plan["dir_renames"]:
        shutil.move(str(src), str(dst))

    entry = plan["registry_entry"]
    if entry is not None:
        entry["name"] = new_name
        entry["updated"] = date.today().isoformat()
        save_registry(plan["registry"], plan["registry_path"])


# ── Rendering ─────────────────────────────────────────────────────


def render_plan(
    plan: dict,
    old_name: str,
    new_name: str,
    scope_label: str,
) -> str:
    """Human-readable diff summary for Claude to show the user."""
    lines = [
        f"Rename {old_name!r} → {new_name!r}  [scope: {scope_label}]",
    ]

    if plan["errors"]:
        lines.append("")
        lines.append("Errors (aborting):")
        lines.extend(f"  - {e}" for e in plan["errors"])
        return "\n".join(lines)

    if plan["warnings"]:
        lines.append("")
        lines.append("Warnings:")
        lines.extend(f"  - {w}" for w in plan["warnings"])

    lines.append("")
    lines.append(f"File edits ({len(plan['file_edits'])}):")
    if plan["file_edits"]:
        for path, _, count in plan["file_edits"]:
            lines.append(f"  {path}  ({count} occurrence{'s' if count != 1 else ''})")
    else:
        lines.append("  (none)")

    lines.append("")
    lines.append(f"Directory renames ({len(plan['dir_renames'])}):")
    for src, dst in plan["dir_renames"]:
        lines.append(f"  {src} → {dst}")

    lines.append("")
    if plan["registry_entry"] is not None:
        lines.append(
            f"Registry: update entry {old_name!r} → {new_name!r} "
            f"in {plan['registry_path']}"
        )
    else:
        lines.append("Registry: no matching entry (would be an error)")

    return "\n".join(lines)


# ── CLI ───────────────────────────────────────────────────────────


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Rename a skill end-to-end.")
    parser.add_argument("old_name")
    parser.add_argument("new_name")
    parser.add_argument(
        "--scope",
        choices=["project", "user"],
        default=None,
        help="Force project or user scope. Auto-detects if omitted.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print the plan without applying it.",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the plan as JSON instead of prose. Implies --dry-run.",
    )
    parser.add_argument(
        "--project-dir",
        type=Path,
        default=None,
        help="Override project root (for tests).",
    )

    args = parser.parse_args(argv)
    project_dir = args.project_dir or Path.cwd()

    skills_root, scope_label = resolve_skills_root(args.scope, project_dir)
    plan = build_plan(args.old_name, args.new_name, skills_root, project_dir)

    if args.json:
        print(json.dumps(
            {
                "scope": scope_label,
                "errors": plan["errors"],
                "warnings": plan["warnings"],
                "file_edits": [[str(p), c] for p, _, c in plan["file_edits"]],
                "dir_renames": [[str(s), str(d)] for s, d in plan["dir_renames"]],
                "registry_path": str(plan["registry_path"]),
            },
            indent=2,
        ))
        return 1 if plan["errors"] else 0

    print(render_plan(plan, args.old_name, args.new_name, scope_label))

    if plan["errors"]:
        return 1

    if args.dry_run:
        print("\n(dry-run — no changes applied)")
        return 0

    execute_plan(plan)
    print(f"\nDone. Renamed {args.old_name!r} → {args.new_name!r}.")
    return 0


if __name__ == "__main__":  # pragma: no cover — entry guard
    sys.exit(main())

"""Real-execution trigger eval.

Spawns `claude -p <query>` subprocesses and parses their stream-json output
to detect whether the Skill tool fires with a target skill name. Adapted
from Anthropic skill-creator's run_eval.py.

Why subprocess + stream parsing (instead of LLM-as-judge):
- A model asked "would this trigger?" confabulates — its answer depends on
  how the question is phrased, not on actual runtime behavior.
- Real runtime observes the actual trigger path: Claude Code's loader sees
  the description in the skills index, decides, and emits a tool_use. That
  is ground truth.
- stream-json + `--include-partial-messages` lets us early-exit the instant
  `content_block_start` announces our skill's tool_use — ~1s signal vs.
  ~10s waiting for the assistant message.

Note on subprocess cwd: this module deliberately runs `claude -p` with
`cwd=project_root` — not `/tmp` as the generation-path rule in
LEARNED.local.md prescribes. Rationale: that rule addresses CLAUDE.md
leaking house style into *generated text*; here the child's textual
output is irrelevant — we only observe whether/when the Skill tool fires.
Running inside project_root is required so Claude Code discovers
`.claude/commands/<slug>.md` and offers the skill as a candidate.
"""

from __future__ import annotations

import json
import os
import select
import subprocess
import time
import uuid
from pathlib import Path

# Defaults match Anthropic skill-creator's run_eval.py. Callers override via
# the `timeout` kwarg; DEFAULT_TIMEOUT exported for consistency.
DEFAULT_TIMEOUT = 30


def find_project_root(start: Path | None = None) -> Path:
    """Walk up from `start` (cwd default) looking for a `.claude/` directory.

    Claude Code uses the same walk to discover its project root. The temp
    command file we write below must land inside *that* `.claude/commands/`,
    or `claude -p` won't see it as a skill candidate. Returns `start` itself
    when no `.claude/` is found upward (caller's own cwd assumption wins).

    Uses os.path.abspath (not Path.resolve) to avoid symlink expansion.
    On macOS, `/tmp` → `/private/tmp` under resolve, which changes the
    recorded project root across processes and breaks anything that
    assumes a stable string path (command file cleanup races, log
    correlation). Shell hooks don't canonicalize either — keeping the
    Python side literal keeps the two aligned.
    """
    start_path = start or Path.cwd()
    current = Path(os.path.abspath(str(start_path)))
    for parent in [current, *current.parents]:
        if (parent / ".claude").is_dir():
            return parent
    return current


def run_single_query(
    query: str,
    skill_name: str,
    description: str,
    *,
    timeout: int = DEFAULT_TIMEOUT,
    project_root: Path | None = None,
    model: str | None = None,
) -> bool:
    """Run `claude -p <query>` and detect whether the skill fired.

    Writes a throwaway command file under `<project_root>/.claude/commands/`
    with a unique `<skill_name>-eval-<uuid>` slug so it appears as a Skill
    candidate without colliding with real commands. Parse stream-json; return
    True at first tool_use input containing our slug — then kill the child.

    Returns False on timeout, unrelated tool firing, or clean completion
    without our slug ever appearing. Temp command file is removed in finally.
    """
    root = project_root or find_project_root()
    unique_id = uuid.uuid4().hex[:8]
    clean_name = f"{skill_name}-eval-{unique_id}"
    commands_dir = root / ".claude" / "commands"
    command_file = commands_dir / f"{clean_name}.md"

    try:
        commands_dir.mkdir(parents=True, exist_ok=True)
        # Normalize line endings before indenting — stray CR in a block-literal
        # YAML value becomes part of the string on strict parsers, corrupting
        # the description the loader reads back.
        normalized = description.replace("\r\n", "\n").replace("\r", "\n")
        # YAML block scalar avoids quote-escaping trouble on arbitrary descriptions.
        indented = "\n  ".join(normalized.split("\n"))
        command_file.write_text(
            f"---\ndescription: |\n  {indented}\n---\n\n"
            f"# {skill_name}\n\nThis skill handles: {normalized}\n"
        )

        cmd = [
            "claude", "-p", query,
            "--output-format", "stream-json",
            "--verbose",
            "--include-partial-messages",
        ]
        if model:
            cmd.extend(["--model", model])

        # CLAUDECODE guards against interactive terminal nesting; programmatic
        # subprocess invocation is fine, so drop the flag to let the child run.
        env = {k: v for k, v in os.environ.items() if k != "CLAUDECODE"}

        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            cwd=str(root),
            env=env,
        )
        return _parse_stream_for_trigger(process, clean_name, timeout)
    except (OSError, FileNotFoundError):
        return False
    finally:
        if command_file.exists():
            try:
                command_file.unlink()
            except OSError:
                pass  # best-effort cleanup; orphaned files are harmless


def _parse_stream_for_trigger(
    process: subprocess.Popen, clean_name: str, timeout: int,
) -> bool:
    """Read stream-json line-by-line, return True on first Skill/Read hit.

    Always terminates the process before returning — orphaned claude children
    would hold file locks on the temp command file.
    """
    triggered = False
    start = time.time()
    buffer = ""
    pending_tool_name: str | None = None
    accumulated_json = ""

    try:
        while time.time() - start < timeout:
            if process.poll() is not None:
                remaining = process.stdout.read() if process.stdout else b""
                if remaining:
                    buffer += remaining.decode("utf-8", errors="replace")
                break

            if not process.stdout:
                break
            ready, _, _ = select.select([process.stdout], [], [], 1.0)
            if not ready:
                continue

            chunk = os.read(process.stdout.fileno(), 8192)
            if not chunk:
                break
            buffer += chunk.decode("utf-8", errors="replace")

            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                line = line.strip()
                if not line:
                    continue
                try:
                    event = json.loads(line)
                except json.JSONDecodeError:
                    continue

                verdict = _handle_event(event, clean_name, pending_tool_name, accumulated_json)
                # verdict is (state_update_or_None, final_bool_or_None)
                state_update, final = verdict
                if state_update is not None:
                    pending_tool_name, accumulated_json = state_update
                if final is not None:
                    return final

                if event.get("type") == "result":
                    return triggered
    finally:
        # Close stdout before wait(): claude can easily write > 64KB of stream
        # events before we kill it. On macOS kernel-pipe buffers that exceed
        # the PIPE_BUF limit make wait() block until the writer side is
        # drained, producing a deadlock. Closing our read end lets the kernel
        # reap the child's write syscalls with EPIPE.
        if process.stdout:
            try:
                process.stdout.close()
            except OSError:
                pass
        if process.poll() is None:
            process.kill()
            process.wait()

    return triggered


def _handle_event(
    event: dict, clean_name: str,
    pending_tool_name: str | None, accumulated_json: str,
) -> tuple[tuple[str | None, str] | None, bool | None]:
    """Classify one stream-json event. Return (state_update, final_verdict).

    state_update: new (pending_tool_name, accumulated_json) if the event
    advances parser state, else None.
    final_verdict: True/False if the event resolves the run, else None.
    """
    # Fast path: stream_event arrives before the full assistant message.
    if event.get("type") == "stream_event":
        se = event.get("event", {})
        se_type = se.get("type", "")
        if se_type == "content_block_start":
            cb = se.get("content_block", {})
            if cb.get("type") == "tool_use":
                tool_name = cb.get("name", "")
                if tool_name in ("Skill", "Read"):
                    return ((tool_name, ""), None)
                # Unrelated tool as the first tool_use -> skill did not
                # trigger on this query. Matches official skill-creator
                # semantics (skill must fire as the first action, else
                # Claude is handling the task without it).
                return (None, False)
        elif se_type == "content_block_delta" and pending_tool_name:
            delta = se.get("delta", {})
            if delta.get("type") == "input_json_delta":
                accumulated_json += delta.get("partial_json", "")
                if clean_name in accumulated_json:
                    return (None, True)
                return ((pending_tool_name, accumulated_json), None)
        elif se_type in ("content_block_stop", "message_stop"):
            if pending_tool_name:
                return (None, clean_name in accumulated_json)
            if se_type == "message_stop":
                return (None, False)
    # Slow path: full assistant message arrived (no partial-stream flag or
    # missed partial). Inspect tool_use content blocks directly.
    elif event.get("type") == "assistant":
        message = event.get("message", {})
        for content in message.get("content", []):
            if content.get("type") != "tool_use":
                continue
            tool_name = content.get("name", "")
            tool_input = content.get("input", {})
            if tool_name == "Skill" and clean_name in tool_input.get("skill", ""):
                return (None, True)
            if tool_name == "Read" and clean_name in tool_input.get("file_path", ""):
                return (None, True)
        return (None, False)
    return (None, None)