fix(sleep): forward sample_id on the tool replay route

CHANGELOG.md

@@ -129,6 +129,12 @@ All notable changes to SkillOpt are documented here. This project adheres to
   value is rejected with a clear error.
 
 ### Fixed
+- Repeated rollouts of a SkillOpt-Sleep tool task are separate samples again.
+  `replay_one` forwarded `sample_id` on the text route only, so with
+  `dream_rollouts > 1` a `tool_called` task on a backend without a real tool
+  loop (Pi, Azure, mock) was answered once and that answer was reused for
+  every rollout. `attempt_with_tools` now accepts `sample_id` on every shipped
+  backend; custom overrides without that parameter keep working.
 - Close four ways invalid evidence could reach the SkillOpt-Sleep gate: recall
   no longer brings a content twin of tonight's held-out task into training,
   a measured tool route no longer credits a self-reported `TOOL_CALL:` marker,

docs/sleep/README.md

@@ -535,6 +535,11 @@ correctness signal; the validation gate still governs what ships.
 | `recall_k` | `0` | Associative recall — pull the K most-similar past tasks (from a persisted archive) into tonight's dream. |
 | `dream_factor` | `0` | Add N lightweight synthetic variants of each task. |
 
+Each of the K rollouts is a separate sample on both replay routes. A task whose
+rule judge requires a tool call takes the tool route; there the rollout index
+is forwarded as `sample_id`, so a backend without a real tool loop (Pi, Azure)
+is asked K times and no longer reuses its first answer for every rollout.
+
 The nightly cycle persists task split exposure separately from the capped recall
 archive. Once a task has entered training, re-mining it or a recorded derivative
 cannot move it into validation or test; validation exposure likewise prevents

skillopt_sleep/backend.py

@@ -46,6 +46,52 @@ def skill_hash(content: str) -> str:
     return hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]
 
 
+def accepts_sample_id(method: Any) -> bool:
+    """True when ``method`` can be called with a ``sample_id`` keyword.
+
+    Backends written before repeated rollouts existed may override
+    ``attempt`` or ``attempt_with_tools`` without that parameter. Callers use
+    this to keep calling such overrides with their original arguments.
+    """
+    import inspect
+
+    try:
+        params = inspect.signature(method).parameters.values()
+    except (TypeError, ValueError):
+        return False
+    return any(
+        param.kind is param.VAR_KEYWORD
+        or (
+            param.name == "sample_id"
+            and param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY)
+        )
+        for param in params
+    )
+
+
+def call_attempt_with_tools(
+    backend: "Backend",
+    task: TaskRecord,
+    skill: str,
+    memory: str,
+    tools: List[str],
+    *,
+    sample_id: int = 0,
+) -> Tuple[str, List[str]]:
+    """Run the tool route, forwarding ``sample_id`` when the backend accepts it.
+
+    Sample zero keeps the historical four-argument call, so existing overrides
+    and cached gate re-scoring are unchanged. An override without a
+    ``sample_id`` parameter is also called with four arguments for every
+    sample; its repeated rollouts are whatever that override returns.
+    """
+    if sample_id and accepts_sample_id(backend.attempt_with_tools):
+        return backend.attempt_with_tools(
+            task, skill, memory, tools, sample_id=sample_id
+        )
+    return backend.attempt_with_tools(task, skill, memory, tools)
+
+
 # ── Backend protocol ──────────────────────────────────────────────────────────
 
 class Backend:
@@ -63,21 +109,35 @@ def attempt(self, task: TaskRecord, skill: str, memory: str,
         raise NotImplementedError
 
     def attempt_with_tools(
-        self, task: TaskRecord, skill: str, memory: str, tools: List[str]
+        self,
+        task: TaskRecord,
+        skill: str,
+        memory: str,
+        tools: List[str],
+        sample_id: int = 0,
     ) -> Tuple[str, List[str]]:
         """Run the task while exposing real tools; return (response, tools_called).
 
         Default: no real tool loop — fall back to plain attempt and let the
         single-shot 'TOOL_CALL: <name>' marker convention surface intent. CLI
         backends override this to expose a genuinely callable tool.
+        ``sample_id`` is forwarded to ``attempt`` so repeated rollouts of a
+        tool task are separate attempts instead of repeats of sample zero.
         """
-        resp = self.attempt(task, skill, memory)
+        resp = self._attempt_sample(task, skill, memory, sample_id)
         called: List[str] = []
         for t in tools:
             if re.search(r"(?i)\btool_call\s*:\s*%s\b" % re.escape(t), resp):
                 called.append(t)
         return resp, called
 
+    def _attempt_sample(
+        self, task: TaskRecord, skill: str, memory: str, sample_id: int
+    ) -> str:
+        if sample_id and accepts_sample_id(self.attempt):
+            return self.attempt(task, skill, memory, sample_id=sample_id)
+        return self.attempt(task, skill, memory)
+
     def judge(self, task: TaskRecord, response: str) -> Tuple[float, float, str]:
         raise NotImplementedError
 
@@ -221,13 +281,13 @@ def attempt(self, task: TaskRecord, skill: str, memory: str,
             return f"approximately {mangled} (format not applied)"
         return "(attempted, no checkable reference)"
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
         # Deterministic tool model: the mock "calls" a tool iff the skill+memory
         # contains an explicit instruction to use it (a learned rule mentioning
         # the tool name or "search"). The deficient skill says NOT to, so
         # baseline calls nothing; a learned "use ./search" rule flips it.
         ctx = ((skill or "") + "\n" + (memory or "")).lower()
-        resp = self.attempt(task, skill, memory)
+        resp = self._attempt_sample(task, skill, memory, sample_id)
         called = []
         for t in (tools or []):
             tl = t.lower()
@@ -931,7 +991,10 @@ def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
         self._detect_cli_error(out, proc.stderr or "")
         return out
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already a separate attempt; ``sample_id`` needs no salt here.
+        del sample_id
         self._reset_call_delta()
         # Expose a REAL, callable `search` tool (a shell shim that logs each
         # call) so the gbrain quick-answerer judge (tool_called=search) is
@@ -1421,7 +1484,11 @@ def attempt_with_tools(
         skill: str,
         memory: str,
         tools: List[str],
+        sample_id: int = 0,
     ) -> Tuple[str, List[str]]:
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already a separate attempt; ``sample_id`` needs no salt here.
+        del sample_id
         self.last_call_error = ""
         self._reset_call_delta()
         if not self.tool_replay:
@@ -1683,7 +1750,10 @@ def _call(self, prompt: str, *, max_tokens: int = 1024, retries: int = 3) -> str
                 _t.sleep(min(6.0, (2 ** attempt) * 0.5) + _r.random() * 0.3)
         return out
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already a separate attempt; ``sample_id`` needs no salt here.
+        del sample_id
         self._reset_call_delta()
         # Codex exec runs in a sandbox with shell access; expose the same real
         # `search` shim and let it run (workspace-write so the shim can log).
@@ -1950,7 +2020,10 @@ def _parse_jsonl_response(raw: str) -> str:
                     parts.append(content)
         return "\n".join(parts).strip()
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already a separate attempt; ``sample_id`` needs no salt here.
+        del sample_id
         self._reset_call_delta()
         # Expose REAL, callable tool shims in the working directory so the
         # gbrain quick-answerer judge (tool_called=search) is validated
@@ -2286,8 +2359,9 @@ def attempt_with_tools(
         skill: str,
         memory: str,
         tools: List[str],
+        sample_id: int = 0,
     ) -> Tuple[str, List[str]]:
-        del task, skill, memory, tools
+        del task, skill, memory, tools, sample_id
         raise self._error(
             "Cursor tool-aware replay is temporarily disabled pending live "
             "Cursor permission-boundary validation"
@@ -2332,9 +2406,11 @@ def attempt(self, task, skill, memory, sample_id: int = 0):
         self._target_tokens_before = self.target.tokens_used()
         return self.target.attempt(task, skill, memory, sample_id=sample_id)
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
         self._target_tokens_before = self.target.tokens_used()
-        return self.target.attempt_with_tools(task, skill, memory, tools)
+        return call_attempt_with_tools(
+            self.target, task, skill, memory, tools, sample_id=sample_id
+        )
 
     def judge(self, task, response):
         # local rule/exact judging needs no model; delegate to target which

skillopt_sleep/replay.py

@@ -12,7 +12,7 @@
 
 from typing import List, Tuple
 
-from skillopt_sleep.backend import Backend
+from skillopt_sleep.backend import Backend, call_attempt_with_tools
 from skillopt_sleep.types import ReplayResult, TaskRecord
 
 
@@ -31,7 +31,8 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
                sample_id: int = 0) -> ReplayResult:
     """``sample_id`` distinguishes repeated dream rollouts of the same
     (task, skill, memory) in the attempt cache — without it all K rollouts
-    collapse to one cached response and the contrastive signal is always 0."""
+    collapse to one cached response and the contrastive signal is always 0.
+    It is forwarded on both the text route and the tool route."""
     import time
     tools = _required_tools(task)
     tools_called: List[str] = []
@@ -42,7 +43,9 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
     token_delta_fn = getattr(backend, "token_delta", None)
     tokens_before = None if token_delta_fn is not None else backend.tokens_used()
     if tools:
-        response, tools_called = backend.attempt_with_tools(task, skill, memory, tools)
+        response, tools_called = call_attempt_with_tools(
+            backend, task, skill, memory, tools, sample_id=sample_id
+        )
     else:
         response = backend.attempt(task, skill, memory, sample_id=sample_id)
     latency_ms = (time.time() - t0) * 1000.0