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 1/3] 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()

From e3437fd7b2420c960d890eb8e288866eaf96af2b Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Tue, 28 Jul 2026 16:29:50 +0000
Subject: [PATCH 2/3] feat(sleep): propagate harvested skill hints into mined
 tasks

Add an optional TaskRecord.skill_hint and a session_skill_hint helper. A session
with exactly one harvested skill names it; absent or ambiguous hints stay empty
so those tasks remain in the existing catch-all path.

Refs #120
---
 skillopt_sleep/llm_miner.py    |   3 +
 skillopt_sleep/mine.py         |  19 +++++
 skillopt_sleep/types.py        |   4 ++
 tests/test_sleep_skill_hint.py | 128 +++++++++++++++++++++++++++++++++
 4 files changed, 154 insertions(+)
 create mode 100644 tests/test_sleep_skill_hint.py

diff --git a/skillopt_sleep/llm_miner.py b/skillopt_sleep/llm_miner.py
index ddf07803..f2d5caf2 100644
--- a/skillopt_sleep/llm_miner.py
+++ b/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,6 +66,7 @@ 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(
@@ -72,6 +74,7 @@ def _mk_task(d: SessionDigest, obj: Dict[str, Any], idx: int) -> TaskRecord | No
             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
 
diff --git a/skillopt_sleep/mine.py b/skillopt_sleep/mine.py
index 9a258274..ffba88e8 100644
--- a/skillopt_sleep/mine.py
+++ b/skillopt_sleep/mine.py
@@ -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,6 +227,12 @@ 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())
diff --git a/skillopt_sleep/types.py b/skillopt_sleep/types.py
index 6d3dc77b..bf559dbf 100644
--- a/skillopt_sleep/types.py
+++ b/skillopt_sleep/types.py
@@ -70,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,
diff --git a/tests/test_sleep_skill_hint.py b/tests/test_sleep_skill_hint.py
new file mode 100644
index 00000000..0138c2e7
--- /dev/null
+++ b/tests/test_sleep_skill_hint.py
@@ -0,0 +1,128 @@
+"""Tests for propagating harvested skill hints into mined tasks (issue #120).
+
+Pure-stdlib (unittest), deterministic, no API key, no third-party deps.
+Run:  python -m pytest tests/test_sleep_skill_hint.py
+"""
+from __future__ import annotations
+
+import unittest
+
+from skillopt_sleep.llm_miner import _mk_task
+from skillopt_sleep.mine import dedup_tasks, heuristic_mine, session_skill_hint
+from skillopt_sleep.types import SessionDigest, TaskRecord
+
+
+def _digest(session_id="s1", skills=(), project="/repo/example", prompt="add a login form"):
+    return SessionDigest(
+        session_id=session_id,
+        project=project,
+        user_prompts=[prompt],
+        assistant_finals=["done"],
+        tools_used=["Skill"] if skills else ["Read"],
+        skills_used=list(skills),
+    )
+
+
+class TestTaskRecordSkillHint(unittest.TestCase):
+    def test_defaults_to_empty_string(self):
+        self.assertEqual(TaskRecord(id="t1", project="/repo/example", intent="x").skill_hint, "")
+
+    def test_serialization_roundtrip(self):
+        task = TaskRecord(
+            id="t1", project="/repo/example", intent="x", skill_hint="example-skill"
+        )
+        self.assertEqual(task.to_dict()["skill_hint"], "example-skill")
+        self.assertEqual(TaskRecord.from_dict(task.to_dict()).skill_hint, "example-skill")
+
+    def test_legacy_payload_without_skill_hint_still_loads(self):
+        legacy = {"id": "t1", "project": "/repo/example", "intent": "x", "split": "val"}
+        task = TaskRecord.from_dict(legacy)
+        self.assertEqual(task.skill_hint, "")
+        self.assertEqual(task.split, "val")
+
+
+class TestSessionSkillHint(unittest.TestCase):
+    def test_single_skill_is_unambiguous(self):
+        self.assertEqual(session_skill_hint(_digest(skills=["example-skill"])), "example-skill")
+
+    def test_no_skill_is_empty(self):
+        self.assertEqual(session_skill_hint(_digest()), "")
+
+    def test_several_skills_are_ambiguous(self):
+        digest = _digest(skills=["example-skill", "second-skill"])
+        self.assertEqual(session_skill_hint(digest), "")
+
+    def test_repeated_single_skill_stays_unambiguous(self):
+        digest = _digest(skills=["example-skill", "example-skill"])
+        self.assertEqual(session_skill_hint(digest), "example-skill")
+
+
+class TestHeuristicMineSkillHint(unittest.TestCase):
+    def test_hint_propagates_from_session(self):
+        tasks = heuristic_mine([_digest(skills=["example-skill"])])
+        self.assertEqual([t.skill_hint for t in tasks], ["example-skill"])
+
+    def test_ambiguous_and_absent_hints_stay_in_catch_all(self):
+        tasks = heuristic_mine([
+            _digest(session_id="s1", skills=["example-skill", "second-skill"]),
+            _digest(session_id="s2", prompt="write the release notes"),
+        ])
+        self.assertEqual(len(tasks), 2)
+        self.assertEqual([t.skill_hint for t in tasks], ["", ""])
+
+    def test_dedup_keeps_agreeing_hint_and_clears_conflicting_one(self):
+        agreed = dedup_tasks([
+            TaskRecord(id="t1", project="/p", intent="x", skill_hint="example-skill",
+                       source_sessions=["s1"]),
+            TaskRecord(id="t1", project="/p", intent="x", skill_hint="example-skill",
+                       source_sessions=["s2"]),
+        ])
+        self.assertEqual([t.skill_hint for t in agreed], ["example-skill"])
+
+        filled = dedup_tasks([
+            TaskRecord(id="t1", project="/p", intent="x", source_sessions=["s1"]),
+            TaskRecord(id="t1", project="/p", intent="x", skill_hint="example-skill",
+                       source_sessions=["s2"]),
+        ])
+        self.assertEqual([t.skill_hint for t in filled], ["example-skill"])
+
+        conflicting = dedup_tasks([
+            TaskRecord(id="t1", project="/p", intent="x", skill_hint="example-skill",
+                       source_sessions=["s1"]),
+            TaskRecord(id="t1", project="/p", intent="x", skill_hint="second-skill",
+                       source_sessions=["s2"]),
+        ])
+        self.assertEqual([t.skill_hint for t in conflicting], [""])
+
+
+class TestLlmMinerSkillHint(unittest.TestCase):
+    def test_rule_task_carries_the_hint(self):
+        task = _mk_task(
+            _digest(skills=["example-skill"]),
+            {"intent": "add a login form", "checks": [{"op": "contains", "arg": "form"}]},
+            0,
+        )
+        self.assertIsNotNone(task)
+        self.assertEqual(task.reference_kind, "rule")
+        self.assertEqual(task.skill_hint, "example-skill")
+
+    def test_rubric_task_carries_the_hint(self):
+        task = _mk_task(
+            _digest(skills=["example-skill"]),
+            {"intent": "add a login form", "rubric": "mentions validation"},
+            0,
+        )
+        self.assertEqual(task.reference_kind, "rubric")
+        self.assertEqual(task.skill_hint, "example-skill")
+
+    def test_ambiguous_session_leaves_hint_empty(self):
+        task = _mk_task(
+            _digest(skills=["example-skill", "second-skill"]),
+            {"intent": "add a login form", "rubric": "mentions validation"},
+            0,
+        )
+        self.assertEqual(task.skill_hint, "")
+
+
+if __name__ == "__main__":
+    unittest.main()

From 74c70576b7af2fd19aa35e97b8280f08d4adb654 Mon Sep 17 00:00:00 2001
From: Dan Baciu <bogdanbaciu@gmail.com>
Date: Thu, 30 Jul 2026 10:46:28 +0400
Subject: [PATCH 3/3] fix(sleep): address skill hint review feedback (#183)

---
 skillopt_sleep/dream.py        |  3 +-
 skillopt_sleep/mine.py         | 14 +++----
 skillopt_sleep/types.py        |  8 ++--
 tests/test_sleep_skill_hint.py | 72 ++++++++++++++++++++++++++++++++++
 4 files changed, 85 insertions(+), 12 deletions(-)

diff --git a/skillopt_sleep/dream.py b/skillopt_sleep/dream.py
index 28ee79c7..7223874c 100644
--- a/skillopt_sleep/dream.py
+++ b/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
 
diff --git a/skillopt_sleep/mine.py b/skillopt_sleep/mine.py
index ffba88e8..9da30703 100644
--- a/skillopt_sleep/mine.py
+++ b/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
@@ -219,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))
@@ -227,14 +230,11 @@ 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
+    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())
 
 
diff --git a/skillopt_sleep/types.py b/skillopt_sleep/types.py
index bf559dbf..de8ef838 100644
--- a/skillopt_sleep/types.py
+++ b/skillopt_sleep/types.py
@@ -70,10 +70,6 @@ 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,
@@ -84,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)
diff --git a/tests/test_sleep_skill_hint.py b/tests/test_sleep_skill_hint.py
index 0138c2e7..e556f090 100644
--- a/tests/test_sleep_skill_hint.py
+++ b/tests/test_sleep_skill_hint.py
@@ -6,7 +6,9 @@
 from __future__ import annotations
 
 import unittest
+from itertools import permutations, product
 
+from skillopt_sleep.dream import dream_augment, recall_similar
 from skillopt_sleep.llm_miner import _mk_task
 from skillopt_sleep.mine import dedup_tasks, heuristic_mine, session_skill_hint
 from skillopt_sleep.types import SessionDigest, TaskRecord
@@ -40,6 +42,17 @@ def test_legacy_payload_without_skill_hint_still_loads(self):
         self.assertEqual(task.skill_hint, "")
         self.assertEqual(task.split, "val")
 
+    def test_existing_positional_fields_keep_their_slots(self):
+        task = TaskRecord(
+            "t1", "/repo/example", "x", "context", "system", "attempt",
+            "success", "exact", "answer", {}, ["tag"], ["s1"],
+            "val", "dream", "source-task",
+        )
+        self.assertEqual(task.split, "val")
+        self.assertEqual(task.origin, "dream")
+        self.assertEqual(task.derived_from, "source-task")
+        self.assertEqual(task.skill_hint, "")
+
 
 class TestSessionSkillHint(unittest.TestCase):
     def test_single_skill_is_unambiguous(self):
@@ -94,6 +107,65 @@ def test_dedup_keeps_agreeing_hint_and_clears_conflicting_one(self):
         ])
         self.assertEqual([t.skill_hint for t in conflicting], [""])
 
+    def test_dedup_conflict_is_permutation_independent(self):
+        sequences = set(permutations(["skill-a", "skill-b", "skill-a"]))
+        sequences.update(permutations(["skill-b", "skill-a", "skill-b"]))
+        for hints in sequences:
+            with self.subTest(hints=hints):
+                tasks = [
+                    TaskRecord(id="t1", project="/p", intent="x", skill_hint=hint,
+                               source_sessions=[f"s{idx}"])
+                    for idx, hint in enumerate(hints)
+                ]
+                self.assertEqual(dedup_tasks(tasks)[0].skill_hint, "")
+
+    def test_dedup_matches_final_non_empty_hint_set_exhaustively(self):
+        for length in range(1, 5):
+            for hints in product(("", "skill-a", "skill-b"), repeat=length):
+                with self.subTest(hints=hints):
+                    distinct = {hint for hint in hints if hint}
+                    expected = next(iter(distinct)) if len(distinct) == 1 else ""
+                    tasks = [
+                        TaskRecord(id="t1", project="/p", intent="x", skill_hint=hint)
+                        for hint in hints
+                    ]
+                    self.assertEqual(dedup_tasks(tasks)[0].skill_hint, expected)
+
+    def test_dedup_ignores_empty_hints_when_resolving_final_set(self):
+        cases = [
+            (("", "skill-a", ""), "skill-a"),
+            (("skill-a", "", "skill-a"), "skill-a"),
+            (("", "", ""), ""),
+            (("skill-a", "", "skill-b"), ""),
+        ]
+        for hints, expected in cases:
+            with self.subTest(hints=hints):
+                tasks = [
+                    TaskRecord(id="t1", project="/p", intent="x", skill_hint=hint)
+                    for hint in hints
+                ]
+                self.assertEqual(dedup_tasks(tasks)[0].skill_hint, expected)
+
+
+class TestDreamAndRecallSkillHint(unittest.TestCase):
+    def test_dream_augment_preserves_hint(self):
+        source = TaskRecord(
+            id="t1", project="/p", intent="add validation", skill_hint="example-skill"
+        )
+        dreamed = dream_augment([source], factor=1)
+        self.assertEqual(len(dreamed), 1)
+        self.assertEqual(dreamed[0].skill_hint, "example-skill")
+
+    def test_recall_similar_preserves_hint(self):
+        new = TaskRecord(id="new", project="/p", intent="add form validation")
+        historical = TaskRecord(
+            id="old", project="/p", intent="improve form validation",
+            skill_hint="example-skill",
+        )
+        recalled = recall_similar([new], [historical], 1)
+        self.assertEqual(len(recalled), 1)
+        self.assertEqual(recalled[0].skill_hint, "example-skill")
+
 
 class TestLlmMinerSkillHint(unittest.TestCase):
     def test_rule_task_carries_the_hint(self):