feat(sleep): harvest Claude Skill tool invocations

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/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

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()