From 7b35092a437d7548745995408190a36ee1825814 Mon Sep 17 00:00:00 2001 From: Bogdan Baciu <270727553+bogdanbaciu21@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:25:34 +0000 Subject: [PATCH] feat(sleep): harvest Claude Skill tool invocations Add a backward-compatible SessionDigest.skills_used field and populate it from well-formed Claude Skill tool-use blocks only. tools_used behavior and legacy digest payload loading are unchanged. Refs #120 --- skillopt_sleep/harvest.py | 31 ++++++ skillopt_sleep/types.py | 4 + tests/test_sleep_skill_harvest.py | 162 ++++++++++++++++++++++++++++++ 3 files changed, 197 insertions(+) create mode 100644 tests/test_sleep_skill_harvest.py diff --git a/skillopt_sleep/harvest.py b/skillopt_sleep/harvest.py index deb8d884..d0c102ca 100644 --- a/skillopt_sleep/harvest.py +++ b/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, diff --git a/skillopt_sleep/types.py b/skillopt_sleep/types.py index c0ed2e5a..6d3dc77b 100644 --- a/skillopt_sleep/types.py +++ b/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 diff --git a/tests/test_sleep_skill_harvest.py b/tests/test_sleep_skill_harvest.py new file mode 100644 index 00000000..e1d57579 --- /dev/null +++ b/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()