SkillOpt/skillopt/sleep/harvest.py

"""SkillOpt-Sleep — Stage 1: harvest.

Read the user's local Claude Code records (read-only) and normalize them
into :class:`SessionDigest` objects.

Sources (verified schema):
  * ~/.claude/history.jsonl        — one JSON/line:
        {"display": <prompt text>, "pastedContents": {...},
         "timestamp": <epoch ms>, "project": <abs path>}
  * ~/.claude/projects/<slug>/<sessionId>.jsonl — one record/line; the
    records we care about have type "user"/"assistant" and carry:
        message{role, content}, cwd, gitBranch, timestamp, sessionId, version

This module performs NO writes and NO network calls.
"""
from __future__ import annotations

import json
import os
from typing import Any, Dict, Iterable, List, Optional

from skillopt.sleep.types import SessionDigest


# Heuristic phrases that signal the user (dis)approving of prior output.
_NEGATIVE_FEEDBACK = (
    "still broken", "still not", "still wrong", "doesn't work", "does not work",
    "not working", "that's wrong", "thats wrong", "incorrect", "wrong",
    "no,", "nope", "fix it", "didn't", "did not", "broken", "error again",
    "still failing", "still fails", "not fixed", "revert", "undo",
    "不对", "还是不对", "还是不行", "不行", "错了", "有问题", "没修好",
)
_POSITIVE_FEEDBACK = (
    "thanks", "thank you", "perfect", "great", "works now", "fixed",
    "that works", "lgtm", "looks good", "nice", "awesome", "correct",
    "完美", "可以了", "好的", "搞定", "对了", "正确", "谢谢",
)


def _iter_jsonl(path: str) -> Iterable[Dict[str, Any]]:
    try:
        with open(path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    yield json.loads(line)
                except Exception:
                    continue
    except (FileNotFoundError, IsADirectoryError, PermissionError):
        return


def _text_from_content(content: Any) -> str:
    """Flatten a message.content (str or list of blocks) into text."""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: List[str] = []
        for b in content:
            if isinstance(b, dict):
                if b.get("type") == "text" and b.get("text"):
                    parts.append(str(b["text"]))
        return "\n".join(parts)
    return ""


def _tool_names_from_content(content: Any) -> List[str]:
    names: List[str] = []
    if isinstance(content, list):
        for b in content:
            if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("name"):
                names.append(str(b["name"]))
    return names


def _detect_feedback(text: str) -> List[str]:
    low = text.lower()
    sig: List[str] = []
    for ph in _NEGATIVE_FEEDBACK:
        if ph in low:
            sig.append("neg:" + ph)
    for ph in _POSITIVE_FEEDBACK:
        if ph in low:
            sig.append("pos:" + ph)
    return sig


def _is_meta_prompt(text: str) -> bool:
    """Skip slash-commands / system noise that aren't real user intents."""
    t = text.strip()
    if not t:
        return True
    if t.startswith("<") and t.endswith(">"):
        return True
    if t.startswith("/") and len(t.split()) <= 3:
        return True
    if t.startswith("[Pasted text") or t.startswith("Caveat:"):
        return True
    return False


def digest_transcript(path: str) -> Optional[SessionDigest]:
    """Build a SessionDigest from one ``<sessionId>.jsonl`` transcript."""
    session_id = os.path.splitext(os.path.basename(path))[0]
    project = ""
    git_branch = ""
    started = ""
    ended = ""
    user_prompts: List[str] = []
    assistant_finals: List[str] = []
    tools: List[str] = []
    files: List[str] = []
    feedback: List[str] = []
    n_user = 0
    n_asst = 0

    for rec in _iter_jsonl(path):
        rtype = rec.get("type")
        ts = rec.get("timestamp")
        if isinstance(ts, str) and ts:
            if not started:
                started = ts
            ended = ts
        if rec.get("cwd") and not project:
            project = str(rec.get("cwd"))
        if rec.get("gitBranch") and not git_branch:
            git_branch = str(rec.get("gitBranch"))
        if rtype == "file-history-snapshot":
            snap = rec.get("snapshot") or rec.get("files") or {}
            if isinstance(snap, dict):
                files.extend([str(k) for k in list(snap.keys())[:20]])
        msg = rec.get("message")
        if not isinstance(msg, dict):
            continue
        role = msg.get("role")
        content = msg.get("content")
        if role == "user":
            text = _text_from_content(content)
            if text and not _is_meta_prompt(text):
                n_user += 1
                user_prompts.append(text.strip())
                feedback.extend(_detect_feedback(text))
        elif role == "assistant":
            n_asst += 1
            tools.extend(_tool_names_from_content(content))
            text = _text_from_content(content)
            if text.strip():
                assistant_finals.append(text.strip())

    if n_user == 0 and n_asst == 0:
        return None

    # de-dup tools/files preserving order
    def _dedup(xs: List[str]) -> List[str]:
        seen = set()
        out = []
        for x in xs:
            if x not in seen:
                seen.add(x)
                out.append(x)
        return out

    return SessionDigest(
        session_id=session_id,
        project=project,
        git_branch=git_branch,
        started_at=started,
        ended_at=ended,
        user_prompts=user_prompts,
        assistant_finals=assistant_finals[-5:],  # last few finals are the useful ones
        tools_used=_dedup(tools),
        files_touched=_dedup(files),
        feedback_signals=feedback,
        n_user_turns=n_user,
        n_assistant_turns=n_asst,
        raw_path=path,
    )


def _project_matches(project: str, scope: Any, invoked: str) -> bool:
    if scope == "all":
        return True
    if isinstance(scope, (list, tuple)):
        return any(os.path.abspath(project) == os.path.abspath(p) for p in scope)
    # "invoked": match the invoked project (or a subdir of it)
    if not invoked:
        return True
    a = os.path.abspath(project)
    b = os.path.abspath(invoked)
    return a == b or a.startswith(b + os.sep) or b.startswith(a + os.sep)


def harvest(
    transcripts_dir: str,
    *,
    scope: Any = "all",
    invoked_project: str = "",
    since_iso: Optional[str] = None,
    limit: int = 0,
) -> List[SessionDigest]:
    """Walk ~/.claude/projects and return digests matching scope/time.

    Parameters
    ----------
    transcripts_dir : str    ~/.claude/projects
    scope : "all" | "invoked" | list[path]
    invoked_project : str    used when scope == "invoked"
    since_iso : str|None      ISO8601; only sessions ending after this are kept
    limit : int               cap number of digests (0 = no cap)
    """
    digests: List[SessionDigest] = []
    if not os.path.isdir(transcripts_dir):
        return digests

    paths: List[str] = []
    for root, _dirs, files in os.walk(transcripts_dir):
        for fn in files:
            if fn.endswith(".jsonl"):
                paths.append(os.path.join(root, fn))
    # newest first by mtime
    paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)

    for p in paths:
        d = digest_transcript(p)
        if d is None:
            continue
        if not _project_matches(d.project or "", scope, invoked_project):
            continue
        if since_iso and d.ended_at and d.ended_at < since_iso:
            continue
        digests.append(d)
        if limit and len(digests) >= limit:
            break
    return digests