feat(sleep): multi-objective reward (accuracy/tokens/latency) + user preferences

- ReplayResult records per-rollout tokens + latency_ms; replay_one measures them
  (approximated from text length when the backend doesn't track tokens, e.g. mock).
- replay.multi_objective_reward(w_acc, w_tokens, w_latency): weighted reward so a
  skill can be optimized to be cheaper/faster, not only more accurate (cost terms
  normalized vs a reference, default = accuracy-only / backward compatible).
- Backend.preferences (free text) injected into reflect as a prior; build_backend
  attaches it (to the optimizer for dual backends). run_gbrain gains --preferences.

3 new tests (multi-objective ordering, preference injection, cost recording).
29 tests pass; mock gates + 3.8/3.12 compile green.

Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>

skillopt/sleep/backend.py

@@ -38,6 +38,8 @@ def skill_hash(content: str) -> str:
 
 class Backend:
     name = "base"
+    # Optional user preferences (free text) injected into reflect as a prior.
+    preferences: str = ""
 
     def attempt(self, task: TaskRecord, skill: str, memory: str) -> str:
         raise NotImplementedError
@@ -381,6 +383,12 @@ class CliBackend(Backend):
                 "\n# Exact criteria the outputs are FAILING (fix these directly)\n"
                 + "\n".join(f"- {_explain(c)}  [{c}, failed {n}x]" for c, n in crit.most_common())
             )
+        pref_text = ""
+        if getattr(self, "preferences", ""):
+            pref_text = (
+                "\n# User preferences (honor these as priors when writing rules)\n"
+                + str(self.preferences).strip()
+            )
         prompt = (
             "You are SkillOpt's optimizer. The agent keeps failing the recurring "
             f"tasks below. Propose at most {edit_budget} bounded edits to the "
@@ -401,7 +409,8 @@ class CliBackend(Backend):
             'Return ONLY a JSON array: '
             '[{"op":"add|replace|delete","content":"<rule>","anchor":"<text to replace/delete, optional>","rationale":"<why>"}].\n\n'
             f"# Current {target}\n{cur_doc}\n"
-            f"{criteria_text}\n\n"
+            f"{criteria_text}\n"
+            f"{pref_text}\n\n"
             f"# Recurring failures\n{fail_text}"
         )
         # Call with one retry: transient non-JSON replies otherwise waste a whole
@@ -756,16 +765,23 @@ def build_backend(
     target_backend: str = "",
     target_model: str = "",
     codex_path: str = "",
+    preferences: str = "",
 ) -> Backend:
     """Build a single or dual backend.
 
     If optimizer_* or target_* are given, returns a DualBackend routing
     attempt->target and reflect/judge->optimizer. Otherwise a single backend
-    from (backend, model).
+    from (backend, model). ``preferences`` (free text) is attached so reflect
+    uses it as a prior (set on the optimizer for dual backends).
     """
     has_split = any([optimizer_backend, optimizer_model, target_backend, target_model])
     if not has_split:
-        return get_backend(backend, model=model, codex_path=codex_path)
+        be = get_backend(backend, model=model, codex_path=codex_path)
+        be.preferences = preferences
+        return be
     tgt = get_backend(target_backend or backend, model=target_model or model, codex_path=codex_path)
     opt = get_backend(optimizer_backend or backend, model=optimizer_model or model, codex_path=codex_path)
-    return DualBackend(target=tgt, optimizer=opt)
+    opt.preferences = preferences  # reflect runs on the optimizer
+    dual = DualBackend(target=tgt, optimizer=opt)
+    dual.preferences = preferences
+    return dual

skillopt/sleep/experiments/run_gbrain.py

@@ -142,6 +142,7 @@ def main(argv=None) -> int:
     ap.add_argument("--budget-tokens", type=int, default=0,
                     help="approx token budget; auto-plans nights x rollouts when set")
     ap.add_argument("--budget-minutes", type=float, default=0.0)
+    ap.add_argument("--preferences", default="", help="free-text user preferences (prior for reflect)")
     ap.add_argument("--limit-replay", type=int, default=0, help="cap #train tasks (cost control)")
     ap.add_argument("--limit-holdout", type=int, default=0, help="cap #val and #test tasks (cost control)")
     ap.add_argument("--json", action="store_true")
@@ -158,7 +159,7 @@ def main(argv=None) -> int:
         backend=args.backend, model=args.model,
         optimizer_backend=args.optimizer_backend, optimizer_model=args.optimizer_model,
         target_backend=args.target_backend, target_model=args.target_model,
-        codex_path=args.codex_path,
+        codex_path=args.codex_path, preferences=args.preferences,
     )
 
     results = []

skillopt/sleep/replay.py

@@ -27,12 +27,20 @@ def _required_tools(task: TaskRecord) -> List[str]:
 
 
 def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str) -> ReplayResult:
+    import time
     tools = _required_tools(task)
     tools_called: List[str] = []
+    t0 = time.time()
+    tok_before = backend.tokens_used()
     if tools:
         response, tools_called = backend.attempt_with_tools(task, skill, memory, tools)
     else:
         response = backend.attempt(task, skill, memory)
+    latency_ms = (time.time() - t0) * 1000.0
+    tokens = max(0, backend.tokens_used() - tok_before)
+    # if the backend doesn't track tokens (e.g. mock), approximate from text length
+    if tokens == 0:
+        tokens = (len(skill) + len(memory) + len(task.intent) + len(response)) // 4
 
     # rule judges may need the detected tool calls; score locally when possible
     if task.reference_kind == "rule" and task.judge:
@@ -50,6 +58,8 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str) -> R
         task_type=(task.tags[0] if task.tags else "task"),
         judge_rationale=rationale,
         tools_called=tools_called,
+        tokens=int(tokens),
+        latency_ms=round(latency_ms, 1),
     )
 
 
@@ -68,3 +78,41 @@ def aggregate_scores(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> Tuple[floa
     hard = sum(r.hard for _t, r in pairs) / len(pairs)
     soft = sum(r.soft for _t, r in pairs) / len(pairs)
     return hard, soft
+
+
+def aggregate_cost(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> Tuple[float, float]:
+    """Mean (tokens, latency_ms) per task — the cost objectives."""
+    if not pairs:
+        return 0.0, 0.0
+    tok = sum(r.tokens for _t, r in pairs) / len(pairs)
+    lat = sum(r.latency_ms for _t, r in pairs) / len(pairs)
+    return tok, lat
+
+
+def multi_objective_reward(
+    pairs: List[Tuple[TaskRecord, ReplayResult]],
+    *,
+    w_acc: float = 1.0,
+    w_tokens: float = 0.0,
+    w_latency: float = 0.0,
+    token_ref: float = 2000.0,
+    latency_ref_ms: float = 15000.0,
+) -> float:
+    """Weighted reward = accuracy↑, tokens↓, latency↓.
+
+    Cost terms are normalized against a reference and clamped to [0,1], so a
+    response at/under the reference cost contributes ~1.0 and an expensive one
+    less. Weights let the user trade off (default = accuracy only, backward
+    compatible).
+    """
+    if not pairs:
+        return 0.0
+    acc, _soft = aggregate_scores(pairs)
+    tok, lat = aggregate_cost(pairs)
+    tok_score = max(0.0, 1.0 - tok / max(1.0, token_ref)) if token_ref else 0.0
+    lat_score = max(0.0, 1.0 - lat / max(1.0, latency_ref_ms)) if latency_ref_ms else 0.0
+    total_w = w_acc + w_tokens + w_latency
+    if total_w <= 0:
+        return acc
+    return (w_acc * acc + w_tokens * tok_score + w_latency * lat_score) / total_w
+

skillopt/sleep/types.py

@@ -95,6 +95,8 @@ class ReplayResult:
     task_type: str = "task"
     judge_rationale: str = ""
     tools_called: List[str] = field(default_factory=list)
+    tokens: int = 0                   # approx tokens this rollout cost (for token objective)
+    latency_ms: float = 0.0           # wall-clock for this rollout (for latency objective)
 
     def to_dict(self) -> Dict[str, Any]:
         return asdict(self)

tests/test_sleep_engine.py

@@ -232,6 +232,51 @@ class TestLlmMiner(unittest.TestCase):
         self.assertEqual(make_llm_miner(EmptyBackend())([digest]), [])
 
 
+class TestMultiObjectiveAndPrefs(unittest.TestCase):
+    def test_multi_objective_reward(self):
+        from skillopt.sleep.replay import multi_objective_reward
+        from skillopt.sleep.types import ReplayResult, TaskRecord
+        t = TaskRecord(id="t", project="/p", intent="x")
+        expensive = [(t, ReplayResult(id="t", hard=1.0, tokens=4000, latency_ms=20000))]
+        cheap = [(t, ReplayResult(id="t", hard=1.0, tokens=200, latency_ms=1000))]
+        self.assertEqual(
+            multi_objective_reward(expensive, w_acc=1, w_tokens=0, w_latency=0),
+            multi_objective_reward(cheap, w_acc=1, w_tokens=0, w_latency=0),
+        )
+        re = multi_objective_reward(expensive, w_acc=1, w_tokens=1, w_latency=1)
+        rc = multi_objective_reward(cheap, w_acc=1, w_tokens=1, w_latency=1)
+        self.assertGreater(rc, re)
+
+    def test_preferences_injected_into_reflect(self):
+        from skillopt.sleep.backend import CliBackend
+        from skillopt.sleep.types import TaskRecord, ReplayResult
+        captured = {}
+
+        class CapBackend(CliBackend):
+            name = "cap"
+            def _call(self, prompt, *, max_tokens=1024):
+                captured["prompt"] = prompt
+                return "[]"
+
+        be = CapBackend()
+        be.preferences = "Prefer concise British English."
+        t = TaskRecord(id="t", project="/p", intent="x", reference_kind="rule",
+                       judge={"checks": [{"op": "contains", "arg": "z"}]})
+        be.reflect([(t, ReplayResult(id="t", hard=0.0, fail_reason="failed: contains=z"))],
+                   [], "skill", "", edit_budget=2, evolve_skill=True, evolve_memory=False)
+        self.assertIn("British English", captured["prompt"])
+
+    def test_replay_records_cost(self):
+        from skillopt.sleep.backend import MockBackend
+        from skillopt.sleep.replay import replay_one
+        from skillopt.sleep.types import TaskRecord
+        t = TaskRecord(id="t", project="/p", intent="hello world",
+                       reference_kind="exact", reference="hi")
+        r = replay_one(MockBackend(), t, "some skill text", "")
+        self.assertGreater(r.tokens, 0)
+        self.assertGreaterEqual(r.latency_ms, 0.0)
+
+
 class TestMultiRolloutAndBudget(unittest.TestCase):
     def test_rolloutset_stats(self):
         from skillopt.sleep.rollout import RolloutSet