feat(sleep): group mined tasks by skill hint with catch-all fallback

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 Any, Callable, Dict, 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:
@@ -214,11 +227,41 @@ def dedup_tasks(tasks: List[TaskRecord]) -> List[TaskRecord]:
             order = {"success": 3, "fail": 2, "mixed": 1, "unknown": 0}
             if order.get(t.outcome, 0) > order.get(ex.outcome, 0):
                 ex.outcome = t.outcome
+            # merged sessions disagreeing about the skill make the hint
+            # ambiguous, so drop back to the catch-all path
+            if not ex.skill_hint:
+                ex.skill_hint = t.skill_hint
+            elif t.skill_hint and t.skill_hint != ex.skill_hint:
+                ex.skill_hint = ""
         else:
             by_id[t.id] = t
     return list(by_id.values())
 
 
+def group_tasks_by_skill_hint(
+    tasks: List[TaskRecord],
+    managed_skill_name: str,
+) -> Dict[str, List[TaskRecord]]:
+    """Group mined tasks by their skill hint, in first-seen order.
+
+    A task ID reaches a hinted group only when every observation of it agrees on
+    the same non-empty hint. Missing hints, conflicting hints, and a mixture of
+    hinted and unhinted observations all fall back to ``managed_skill_name`` —
+    the existing catch-all skill — so ambiguous evidence never invents a group.
+    Each task ID is emitted exactly once, with ``dedup_tasks`` merge semantics.
+    """
+    observed: dict = {}
+    for t in tasks:
+        observed.setdefault(t.id, set()).add((t.skill_hint or "").strip())
+
+    groups: Dict[str, List[TaskRecord]] = {}
+    for task in dedup_tasks(tasks):
+        hints = observed[task.id]
+        hint = next(iter(hints)) if len(hints) == 1 else ""
+        groups.setdefault(hint or managed_skill_name, []).append(task)
+    return groups
+
+
 def assign_splits(
     tasks: List[TaskRecord],
     *,

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
@@ -66,6 +70,10 @@ class TaskRecord:
     judge: Dict[str, Any] = field(default_factory=dict)  # gbrain-style rule judge
     tags: List[str] = field(default_factory=list)
     source_sessions: List[str] = field(default_factory=list)
+    # 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.
+    skill_hint: str = ""
     # split ∈ {train, val, test}.  val + test come ONLY from real mined tasks and
     # never overlap (val gates updates, test is the final held-out measure). train
     # may be dream-augmented (see origin).  Legacy values replay->train,

tests/test_sleep_engine.py

@@ -19,7 +19,13 @@
 from skillopt_sleep.experiments.personas import programmer_persona, researcher_persona
 from skillopt_sleep.harvest import _detect_feedback, _is_meta_prompt, digest_transcript
 from skillopt_sleep.memory import apply_edits, current_learned_lines, extract_learned, set_learned
-from skillopt_sleep.mine import assign_splits, filter_tasks_for_target, heuristic_mine, mine
+from skillopt_sleep.mine import (
+    assign_splits,
+    filter_tasks_for_target,
+    group_tasks_by_skill_hint,
+    heuristic_mine,
+    mine,
+)
 from skillopt_sleep.staging import adopt
 from skillopt_sleep.types import EditRecord, SessionDigest, SleepReport, TaskRecord
 
@@ -2252,5 +2258,81 @@ def _fake_once(prompt, *, max_tokens=1024):
         self.assertIn("REDACTED", joined)
 
 
+class TestGroupTasksBySkillHint(unittest.TestCase):
+    MANAGED = "managed-skill"
+
+    def _task(self, tid, hint="", session="s1", outcome="unknown"):
+        return TaskRecord(
+            id=tid, project="/repo/example", intent="intent " + tid,
+            skill_hint=hint, source_sessions=[session], outcome=outcome,
+        )
+
+    def _ids(self, groups):
+        return {name: [t.id for t in tasks] for name, tasks in groups.items()}
+
+    def test_group_tasks_by_skill_hint_empty_input(self):
+        self.assertEqual(group_tasks_by_skill_hint([], self.MANAGED), {})
+
+    def test_group_tasks_by_skill_hint_legacy_tasks_go_to_managed_skill(self):
+        groups = group_tasks_by_skill_hint(
+            [self._task("t1"), self._task("t2")], self.MANAGED
+        )
+        self.assertEqual(self._ids(groups), {self.MANAGED: ["t1", "t2"]})
+
+    def test_group_tasks_by_skill_hint_blank_hint_goes_to_managed_skill(self):
+        groups = group_tasks_by_skill_hint([self._task("t1", "   ")], self.MANAGED)
+        self.assertEqual(self._ids(groups), {self.MANAGED: ["t1"]})
+
+    def test_group_tasks_by_skill_hint_preserves_first_seen_order(self):
+        groups = group_tasks_by_skill_hint(
+            [
+                self._task("t1", "alpha"),
+                self._task("t2", "beta"),
+                self._task("t3", "alpha"),
+                self._task("t4"),
+            ],
+            self.MANAGED,
+        )
+        self.assertEqual(list(groups), ["alpha", "beta", self.MANAGED])
+        self.assertEqual(
+            self._ids(groups),
+            {"alpha": ["t1", "t3"], "beta": ["t2"], self.MANAGED: ["t4"]},
+        )
+
+    def test_group_tasks_by_skill_hint_merges_duplicate_ids_once(self):
+        groups = group_tasks_by_skill_hint(
+            [
+                self._task("t1", "alpha", session="s1"),
+                self._task("t2", "beta"),
+                self._task("t1", "alpha", session="s2", outcome="success"),
+            ],
+            self.MANAGED,
+        )
+        self.assertEqual(self._ids(groups), {"alpha": ["t1"], "beta": ["t2"]})
+        merged = groups["alpha"][0]
+        self.assertEqual(merged.source_sessions, ["s1", "s2"])
+        self.assertEqual(merged.outcome, "success")
+
+    def test_group_tasks_by_skill_hint_conflicting_hints_go_to_managed_skill(self):
+        groups = group_tasks_by_skill_hint(
+            [self._task("t1", "alpha"), self._task("t1", "beta", session="s2")],
+            self.MANAGED,
+        )
+        self.assertEqual(self._ids(groups), {self.MANAGED: ["t1"]})
+
+    def test_group_tasks_by_skill_hint_partial_hint_evidence_goes_to_managed_skill(self):
+        groups = group_tasks_by_skill_hint(
+            [self._task("t1", "alpha"), self._task("t1", session="s2")],
+            self.MANAGED,
+        )
+        self.assertEqual(self._ids(groups), {self.MANAGED: ["t1"]})
+
+    def test_group_tasks_by_skill_hint_hint_equal_to_managed_skill_is_one_group(self):
+        groups = group_tasks_by_skill_hint(
+            [self._task("t1", self.MANAGED), self._task("t2")], self.MANAGED
+        )
+        self.assertEqual(self._ids(groups), {self.MANAGED: ["t1", "t2"]})
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)