fix(sleep): avoid mutating tasks while grouping

skillopt_sleep/mine.py

@@ -18,6 +18,7 @@
 import os
 import re
 from collections import Counter
+from dataclasses import replace
 from typing import Callable, Dict, List, Optional, Set, Tuple
 
 from skillopt_sleep.backend import CursorBackendError
@@ -254,10 +255,16 @@ def group_tasks_by_skill_hint(
     for t in tasks:
         observed.setdefault(t.id, set()).add((t.skill_hint or "").strip())
 
+    # ``dedup_tasks`` merges records in place, so operate on shallow dataclass
+    # copies (including the only list it mutates) rather than caller-owned tasks.
+    copied = [replace(t, source_sessions=list(t.source_sessions)) for t in tasks]
     groups: Dict[str, List[TaskRecord]] = {}
-    for task in dedup_tasks(tasks):
+    for task in dedup_tasks(copied):
         hints = observed[task.id]
         hint = next(iter(hints)) if len(hints) == 1 else ""
+        # Keep the returned record aligned with the normalized evidence used for
+        # routing. In particular, blank and partially observed hints stay empty.
+        task.skill_hint = hint
         groups.setdefault(hint or managed_skill_name, []).append(task)
     return groups
 

tests/test_sleep_engine.py

@@ -2282,6 +2282,7 @@ def test_group_tasks_by_skill_hint_legacy_tasks_go_to_managed_skill(self):
     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"]})
+        self.assertEqual(groups[self.MANAGED][0].skill_hint, "")
 
     def test_group_tasks_by_skill_hint_preserves_first_seen_order(self):
         groups = group_tasks_by_skill_hint(
@@ -2300,18 +2301,35 @@ def test_group_tasks_by_skill_hint_preserves_first_seen_order(self):
         )
 
     def test_group_tasks_by_skill_hint_merges_duplicate_ids_once(self):
+        first = self._task("t1", "alpha", session="s1")
+        duplicate = self._task("t1", "alpha", session="s2", outcome="success")
         groups = group_tasks_by_skill_hint(
             [
-                self._task("t1", "alpha", session="s1"),
+                first,
                 self._task("t2", "beta"),
-                self._task("t1", "alpha", session="s2", outcome="success"),
+                duplicate,
             ],
             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")
+        self.assertIsNot(merged, first)
+        self.assertEqual(first.source_sessions, ["s1"])
+        self.assertEqual(first.outcome, "unknown")
+        self.assertEqual(duplicate.source_sessions, ["s2"])
+
+    def test_group_tasks_by_skill_hint_normalizes_without_mutating_inputs(self):
+        first = self._task("t1", " alpha ", session="s1")
+        duplicate = self._task("t1", "alpha", session="s2")
+
+        groups = group_tasks_by_skill_hint([first, duplicate], self.MANAGED)
+
+        self.assertEqual(self._ids(groups), {"alpha": ["t1"]})
+        self.assertEqual(groups["alpha"][0].skill_hint, "alpha")
+        self.assertEqual(first.skill_hint, " alpha ")
+        self.assertEqual(duplicate.skill_hint, "alpha")
 
     def test_group_tasks_by_skill_hint_conflicting_hints_go_to_managed_skill(self):
         groups = group_tasks_by_skill_hint(
@@ -2326,6 +2344,7 @@ def test_group_tasks_by_skill_hint_partial_hint_evidence_goes_to_managed_skill(s
             self.MANAGED,
         )
         self.assertEqual(self._ids(groups), {self.MANAGED: ["t1"]})
+        self.assertEqual(groups[self.MANAGED][0].skill_hint, "")
 
     def test_group_tasks_by_skill_hint_hint_equal_to_managed_skill_is_one_group(self):
         groups = group_tasks_by_skill_hint(