feat(sleep): propagate harvested skill hints into mined tasks

skillopt_sleep/dream.py

@@ -22,7 +22,6 @@
 from skillopt_sleep.consolidate import ConsolidationResult, consolidate
 from skillopt_sleep.types import TaskRecord
 
-
 # ── synthetic augmentation ("dream up" variants of today's tasks) ─────────────
 
 _WRAPPERS = [
@@ -49,6 +48,7 @@ def dream_augment(real_tasks: List[TaskRecord], *, factor: int = 1) -> List[Task
                 judge=dict(t.judge), system=t.system,
                 tags=list(t.tags) + ["dream"], split="train",
                 origin="dream", derived_from=t.id,
+                skill_hint=t.skill_hint,
             ))
     return out
 
@@ -90,6 +90,7 @@ def recall_similar(new_tasks: List[TaskRecord], history: List[TaskRecord],
             reference=h.reference, judge=dict(h.judge), system=h.system,
             tags=list(h.tags) + ["recall"], split="train", origin="real",
             derived_from=h.id,
+            skill_hint=h.skill_hint,
         ))
     return out
 

skillopt_sleep/harvest.py

@@ -86,6 +86,34 @@ def _tool_names_from_content(content: Any) -> List[str]:
     return names
 
 
+def _skill_names_from_content(content: Any) -> List[str]:
+    """Extract skill targets from Claude ``Skill`` tool-use blocks.
+
+    Only well-formed invocations count: the block type must be ``tool_use``,
+    its name exactly ``Skill``, and ``input.skill`` a non-blank string. The
+    name is returned whitespace-trimmed but otherwise verbatim; no other tool
+    input and no tool output is read.
+    """
+    names: List[str] = []
+    if not isinstance(content, list):
+        return names
+    for b in content:
+        if not isinstance(b, dict):
+            continue
+        if b.get("type") != "tool_use" or b.get("name") != "Skill":
+            continue
+        args = b.get("input")
+        if not isinstance(args, dict):
+            continue
+        skill = args.get("skill")
+        if not isinstance(skill, str):
+            continue
+        skill = skill.strip()
+        if skill:
+            names.append(skill)
+    return names
+
+
 def _detect_feedback(text: str) -> List[str]:
     low = text.lower()
     sig: List[str] = []
@@ -206,6 +234,7 @@ def digest_transcript(path: str) -> Optional[SessionDigest]:
     user_prompts: List[str] = []
     assistant_finals: List[str] = []
     tools: List[str] = []
+    skills: List[str] = []
     files: List[str] = []
     feedback: List[str] = []
     n_user = 0
@@ -240,6 +269,7 @@ def digest_transcript(path: str) -> Optional[SessionDigest]:
         elif role == "assistant":
             n_asst += 1
             tools.extend(_tool_names_from_content(content))
+            skills.extend(_skill_names_from_content(content))
             text = _text_from_content(content)
             if text.strip():
                 assistant_finals.append(text.strip())
@@ -266,6 +296,7 @@ def _dedup(xs: List[str]) -> List[str]:
         user_prompts=user_prompts,
         assistant_finals=assistant_finals[-5:],  # last few finals are the useful ones
         tools_used=_dedup(tools),
+        skills_used=_dedup(skills),
         files_touched=_dedup(files),
         feedback_signals=feedback,
         n_user_turns=n_user,

skillopt_sleep/llm_miner.py

@@ -24,6 +24,7 @@
 
 from skillopt_sleep import prompts as prompt_registry
 from skillopt_sleep.backend import Backend, _extract_json
+from skillopt_sleep.mine import session_skill_hint
 from skillopt_sleep.types import SessionDigest, TaskRecord
 
 
@@ -65,13 +66,15 @@ def _mk_task(d: SessionDigest, obj: Dict[str, Any], idx: int) -> TaskRecord | No
             reference_kind="rule", judge={"kind": "rule", "checks": clean_checks},
             outcome="success" if satisfied else "fail",
             tags=["mined:llm"], source_sessions=[d.session_id],
+            skill_hint=session_skill_hint(d),
         )
     if rubric:
         return TaskRecord(
             id=tid, project=d.project, intent=intent,
             reference_kind="rubric", reference=rubric,
             outcome="success" if satisfied else "fail",
             tags=["mined:llm"], source_sessions=[d.session_id],
+            skill_hint=session_skill_hint(d),
         )
     return None  # not checkable -> drop
 

skillopt_sleep/mine.py

@@ -18,7 +18,7 @@
 import os
 import re
 from collections import Counter
-from typing import Any, Callable, List, Optional, Set, Tuple
+from typing import Callable, List, Optional, Set, Tuple
 
 from skillopt_sleep.backend import CursorBackendError
 from skillopt_sleep.types import SessionDigest, TaskRecord
@@ -34,6 +34,18 @@ def _short(text: str, n: int = 600) -> str:
     return text if len(text) <= n else text[:n] + " …"
 
 
+def session_skill_hint(digest: SessionDigest) -> str:
+    """Return the session's unambiguous skill, or "" when there is none.
+
+    A session that invoked exactly one skill names the skill its tasks
+    exercised. Zero skills (unknown) and several skills (ambiguous) both
+    return "", which leaves those tasks in the existing catch-all path.
+    """
+    skills = [s for s in (digest.skills_used or []) if s]
+    unique = list(dict.fromkeys(skills))
+    return unique[0] if len(unique) == 1 else ""
+
+
 def _looks_negative(signals: List[str]) -> bool:
     return any(s.startswith("neg:") for s in signals)
 
@@ -196,6 +208,7 @@ def heuristic_mine(
                 reference="",
                 tags=tags,
                 source_sessions=[d.session_id],
+                skill_hint=session_skill_hint(d),
             )
         )
         if len(tasks) >= max_tasks:
@@ -206,7 +219,10 @@ def heuristic_mine(
 def dedup_tasks(tasks: List[TaskRecord]) -> List[TaskRecord]:
     """Merge tasks sharing an id (same project+intent across sessions)."""
     by_id: dict = {}
+    hints_by_id: dict = {}
     for t in tasks:
+        if t.skill_hint:
+            hints_by_id.setdefault(t.id, set()).add(t.skill_hint)
         if t.id in by_id:
             ex = by_id[t.id]
             ex.source_sessions = list(dict.fromkeys(ex.source_sessions + t.source_sessions))
@@ -216,6 +232,9 @@ def dedup_tasks(tasks: List[TaskRecord]) -> List[TaskRecord]:
                 ex.outcome = t.outcome
         else:
             by_id[t.id] = t
+    for task_id, task in by_id.items():
+        hints = hints_by_id.get(task_id, set())
+        task.skill_hint = next(iter(hints)) if len(hints) == 1 else ""
     return list(by_id.values())
 
 

skillopt_sleep/types.py

@@ -29,6 +29,10 @@ class SessionDigest:
     user_prompts: List[str] = field(default_factory=list)
     assistant_finals: List[str] = field(default_factory=list)
     tools_used: List[str] = field(default_factory=list)
+    # Skill targets invoked through the Claude ``Skill`` tool, in first-seen
+    # order. Optional and default-empty: harvesters that do not observe skill
+    # invocations, and digests persisted before this field existed, leave it [].
+    skills_used: List[str] = field(default_factory=list)
     files_touched: List[str] = field(default_factory=list)
     feedback_signals: List[str] = field(default_factory=list)  # "still broken", "perfect", ...
     n_user_turns: int = 0
@@ -76,6 +80,10 @@ class TaskRecord:
     # allowed into val/test, which is the anti-overfitting guarantee.
     origin: str = "real"
     derived_from: str = ""            # for dream tasks: the real task id it varies
+    # Which skill this task exercised, when the source session invoked exactly
+    # one. Empty means unknown or ambiguous, which keeps the task in the
+    # existing managed-skill catch-all path. Kept last for positional compatibility.
+    skill_hint: str = ""
 
     def to_dict(self) -> Dict[str, Any]:
         return asdict(self)

tests/test_sleep_skill_harvest.py

@@ -0,0 +1,162 @@
+"""Tests for Claude ``Skill`` tool-use harvesting (issue #120).
+
+Pure-stdlib (unittest), deterministic, no API key, no third-party deps.
+Run:  python -m pytest tests/test_sleep_skill_harvest.py
+"""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+
+from skillopt_sleep.harvest import digest_transcript
+from skillopt_sleep.types import SessionDigest
+
+
+def _skill_block(skill, name="Skill", block_type="tool_use", extra=None):
+    args = {"skill": skill} if skill is not None else {}
+    if extra:
+        args.update(extra)
+    return {"type": block_type, "name": name, "input": args}
+
+
+def _assistant(content):
+    return {
+        "type": "assistant",
+        "timestamp": "2026-07-28T10:00:00Z",
+        "cwd": "/repo/example",
+        "gitBranch": "main",
+        "message": {"role": "assistant", "content": content},
+    }
+
+
+def _user(text):
+    return {
+        "type": "user",
+        "timestamp": "2026-07-28T09:59:00Z",
+        "cwd": "/repo/example",
+        "message": {"role": "user", "content": text},
+    }
+
+
+class TestSessionDigestSkillsField(unittest.TestCase):
+    def test_defaults_to_empty_list(self):
+        digest = SessionDigest(session_id="s1", project="/repo/example")
+        self.assertEqual(digest.skills_used, [])
+
+    def test_to_dict_includes_empty_skills_used(self):
+        digest = SessionDigest(session_id="s1", project="/repo/example")
+        self.assertIn("skills_used", digest.to_dict())
+        self.assertEqual(digest.to_dict()["skills_used"], [])
+
+    def test_legacy_payload_without_skills_used_still_loads(self):
+        legacy = {"session_id": "s1", "project": "/repo/example", "tools_used": ["Bash"]}
+        known = set(SessionDigest.__dataclass_fields__)
+        digest = SessionDigest(**{k: v for k, v in legacy.items() if k in known})
+        self.assertEqual(digest.skills_used, [])
+        self.assertEqual(digest.tools_used, ["Bash"])
+
+    def test_unknown_key_filter_can_ignore_new_field(self):
+        payload = SessionDigest(
+            session_id="s1", project="/repo/example", skills_used=["example-skill"]
+        ).to_dict()
+        older_known = {
+            f for f in SessionDigest.__dataclass_fields__ if f != "skills_used"
+        }
+        digest = SessionDigest(**{k: v for k, v in payload.items() if k in older_known})
+        self.assertEqual(digest.skills_used, [])
+
+
+class TestSkillHarvest(unittest.TestCase):
+    def _digest(self, records):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "session-example.jsonl")
+            with open(path, "w", encoding="utf-8") as f:
+                for record in records:
+                    f.write(json.dumps(record) + "\n")
+            return digest_transcript(path)
+
+    def test_harvests_skill_invocation(self):
+        digest = self._digest([
+            _user("run the example skill"),
+            _assistant([_skill_block("example-skill")]),
+        ])
+        self.assertIsNotNone(digest)
+        self.assertEqual(digest.skills_used, ["example-skill"])
+        # tools_used keeps its existing meaning, including the Skill tool itself
+        self.assertEqual(digest.tools_used, ["Skill"])
+
+    def test_duplicates_collapse_in_first_seen_order(self):
+        digest = self._digest([
+            _user("run both skills"),
+            _assistant([
+                _skill_block("second-skill"),
+                _skill_block("example-skill"),
+                _skill_block("second-skill"),
+            ]),
+        ])
+        self.assertEqual(digest.skills_used, ["second-skill", "example-skill"])
+
+    def test_whitespace_trimmed_without_rewriting_name(self):
+        digest = self._digest([
+            _user("run it"),
+            _assistant([_skill_block("  Example-Skill:v2  ")]),
+        ])
+        self.assertEqual(digest.skills_used, ["Example-Skill:v2"])
+
+    def test_ordinary_tools_never_populate_skills_used(self):
+        digest = self._digest([
+            _user("read the file"),
+            _assistant([
+                {"type": "tool_use", "name": "Read", "input": {"file_path": "/repo/a.py"}},
+                {"type": "text", "text": "I used the Skill tool description here"},
+            ]),
+        ])
+        self.assertEqual(digest.skills_used, [])
+        self.assertEqual(digest.tools_used, ["Read"])
+
+    def test_malformed_blocks_are_ignored(self):
+        digest = self._digest([
+            _user("run it"),
+            _assistant([
+                _skill_block("lowercase-tool", name="skill"),
+                _skill_block("wrong-type", block_type="tool_result"),
+                {"type": "tool_use", "name": "Skill"},
+                {"type": "tool_use", "name": "Skill", "input": "example-skill"},
+                {"type": "tool_use", "name": "Skill", "input": {"skill": 7}},
+                _skill_block("   "),
+                _skill_block(None),
+                "not-a-block",
+            ]),
+        ])
+        self.assertEqual(digest.skills_used, [])
+
+    def test_malformed_jsonl_record_is_non_fatal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "session-example.jsonl")
+            with open(path, "w", encoding="utf-8") as f:
+                f.write(json.dumps(_user("run it")) + "\n")
+                f.write("{not json\n")
+                f.write("\n")
+                f.write(json.dumps(_assistant([_skill_block("example-skill")])) + "\n")
+            digest = digest_transcript(path)
+        self.assertIsNotNone(digest)
+        self.assertEqual(digest.skills_used, ["example-skill"])
+
+    def test_other_tool_arguments_and_outputs_stay_out_of_the_digest(self):
+        digest = self._digest([
+            _user("run it"),
+            _assistant([
+                _skill_block("example-skill", extra={"secret_arg": "do-not-copy"}),
+                {"type": "tool_result", "content": "output should not copy"},
+            ]),
+        ])
+        serialized = json.dumps(digest.to_dict())
+        self.assertEqual(digest.skills_used, ["example-skill"])
+        self.assertNotIn("do-not-copy", serialized)
+        self.assertNotIn("output should not copy", serialized)
+
+
+if __name__ == "__main__":
+    unittest.main()