From 492e1fa72300a52faa67159db4fa4c59711a0fd4 Mon Sep 17 00:00:00 2001 From: Dan Baciu Date: Wed, 11 Nov 2026 14:35:53 +0000 Subject: [PATCH] fix(sleep): forward sample_id on the tool replay route replay_one passed sample_id to attempt on the text route only. On the tool route it called attempt_with_tools without it, and the inherited marker fallback (used by Pi, Azure and any backend without a real tool loop) called attempt at sample zero. Repeated rollouts of a tool_called task were therefore answered once and served from that one attempt-cache entry: with dream_rollouts=4, a tool task through Pi made one provider call and reported four identical scores, so contrastive reflection saw no spread on tool tasks. - attempt_with_tools accepts sample_id on every shipped backend; the inherited fallback and the mock tool model forward it to attempt, the real tool loops already start a fresh uncached process per call, and DualBackend forwards it to the target - replay_one forwards sample_id on the tool route; sample zero keeps the four-argument call, so the default gate path is unchanged - a custom attempt or attempt_with_tools override without a sample_id parameter is still called with its original arguments --- CHANGELOG.md | 6 + docs/sleep/README.md | 5 + skillopt_sleep/backend.py | 96 ++++++++++++-- skillopt_sleep/replay.py | 9 +- tests/test_tool_route_sample_id.py | 194 +++++++++++++++++++++++++++++ 5 files changed, 297 insertions(+), 13 deletions(-) create mode 100644 tests/test_tool_route_sample_id.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ef83812d..e7edff96 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/docs/sleep/README.md b/docs/sleep/README.md index eec3e6cd..d11c99da 100644 --- a/docs/sleep/README.md +++ b/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 diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py index 9342c7be..4f2ac8db 100644 --- a/skillopt_sleep/backend.py +++ b/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: ' 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 diff --git a/skillopt_sleep/replay.py b/skillopt_sleep/replay.py index a73709c0..c7f7261b 100644 --- a/skillopt_sleep/replay.py +++ b/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 diff --git a/tests/test_tool_route_sample_id.py b/tests/test_tool_route_sample_id.py new file mode 100644 index 00000000..ca5e3e56 --- /dev/null +++ b/tests/test_tool_route_sample_id.py @@ -0,0 +1,194 @@ +"""Repeated rollouts of a tool task must reach the backend as separate samples. + +``replay_one`` forwarded ``sample_id`` on the text route only. On the tool +route it called ``attempt_with_tools`` without it, and the inherited marker +fallback called ``attempt`` at sample zero, so K rollouts of a ``tool_called`` +task were served from one attempt-cache entry. + +Only the provider child process (``subprocess.run``) is faked. No model is +called. +""" +from __future__ import annotations + +import inspect +import itertools +import subprocess +from unittest import mock + +import pytest + +from skillopt_sleep import backend as backend_module +from skillopt_sleep.backend import DualBackend, MockBackend, PiCliBackend +from skillopt_sleep.replay import replay_one +from skillopt_sleep.rollout import multi_rollout +from skillopt_sleep.types import TaskRecord + +_TOOL_CHECK = {"op": "tool_called", "arg": "search"} +_TEXT_CHECK = {"op": "contains", "arg": "TOOL_CALL"} + + +def _rule_task(check: dict, task_id: str = "tool-task") -> TaskRecord: + return TaskRecord( + id=task_id, + project="/project", + intent="Please look it up", + reference_kind="rule", + judge={"kind": "rule", "checks": [check]}, + ) + + +def _alternating_provider(): + """Fake Pi child: calls the tool on even calls, answers without it on odd.""" + counter = itertools.count() + + def run(cmd, *, input, **kwargs): + if next(counter) % 2 == 0: + out = "TOOL_CALL: search\nanswer" + else: + out = "answer without the tool" + return subprocess.CompletedProcess(cmd, 0, stdout=out, stderr="") + + return run + + +@pytest.mark.parametrize( + "check", [_TOOL_CHECK, _TEXT_CHECK], ids=["tool-route", "text-route-control"] +) +def test_dream_rollouts_reach_the_provider_as_separate_samples(check) -> None: + """Four rollouts are four provider calls on both routes. + + The text route already behaved this way and is the control. The tool route + made one provider call and reported four identical scores. + """ + with mock.patch( + "skillopt_sleep.backend.subprocess.run", side_effect=_alternating_provider() + ) as run: + rollout_set = multi_rollout( + PiCliBackend(), _rule_task(check), "skill", "", k=4, workers=1 + ) + assert run.call_count == 4 + assert [attempt.hard for attempt in rollout_set.attempts] == [1.0, 0.0, 1.0, 0.0] + + +def test_inherited_tool_fallback_keys_the_attempt_cache_by_sample_id() -> None: + """Three distinct sample ids are three provider calls; a repeated id is a + cache hit, so re-scoring an already-sampled rollout stays free.""" + backend = PiCliBackend() + + def run(cmd, *, input, **kwargs): + return subprocess.CompletedProcess( + cmd, 0, stdout="TOOL_CALL: search\nanswer", stderr="" + ) + + with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=run) as run_mock: + for sample_id in (0, 1, 2, 0, 1): + result = replay_one( + backend, _rule_task(_TOOL_CHECK), "skill", "", sample_id=sample_id + ) + assert result.tools_called == ["search"] + assert result.hard == 1.0 + assert run_mock.call_count == 3 + + +def test_mock_tool_model_forwards_sample_id_to_attempt() -> None: + seen: list[int] = [] + + class _Recording(MockBackend): + def attempt(self, task, skill, memory, sample_id=0): + seen.append(sample_id) + return "answer" + + backend = _Recording() + for sample_id in range(3): + replay_one(backend, _rule_task(_TOOL_CHECK), "use search", "", sample_id=sample_id) + assert seen == [0, 1, 2] + + +def test_dual_backend_forwards_tool_sample_ids_to_the_target() -> None: + seen: list[int] = [] + + class _Target(MockBackend): + def attempt(self, task, skill, memory, sample_id=0): + seen.append(sample_id) + return "TOOL_CALL: search" + + dual = DualBackend(_Target(), MockBackend()) + for sample_id in range(3): + replay_one(dual, _rule_task(_TOOL_CHECK), "", "", sample_id=sample_id) + assert seen == [0, 1, 2] + + +def test_sample_zero_keeps_the_four_argument_tool_call() -> None: + """Sample zero passes no ``sample_id`` keyword, so the default gate path + calls an override exactly as it did before; later samples pass it.""" + calls: list[dict] = [] + + class _Recording(MockBackend): + def attempt_with_tools(self, task, skill, memory, tools, **kwargs): + calls.append(dict(kwargs)) + return "answer", list(tools) + + backend = _Recording() + for sample_id in (0, 2): + replay_one(backend, _rule_task(_TOOL_CHECK), "", "", sample_id=sample_id) + assert calls == [{}, {"sample_id": 2}] + + +def test_tool_override_without_sample_id_parameter_is_still_called() -> None: + """A custom backend whose tool override predates ``sample_id`` keeps + working for every sample. It is called with its original four arguments, + so its repeated rollouts are whatever that override returns.""" + calls: list[int] = [] + + class _Legacy(MockBackend): + def attempt_with_tools(self, task, skill, memory, tools): + calls.append(len(calls)) + return "answer", list(tools) + + backend = _Legacy() + for sample_id in range(3): + result = replay_one( + backend, _rule_task(_TOOL_CHECK), "", "", sample_id=sample_id + ) + assert result.hard == 1.0 + assert calls == [0, 1, 2] + + dual = DualBackend(_Legacy(), MockBackend()) + assert replay_one(dual, _rule_task(_TOOL_CHECK), "", "", sample_id=1).hard == 1.0 + + +def test_attempt_override_without_sample_id_parameter_is_still_called() -> None: + """The inherited tool fallback does not pass ``sample_id`` to an + ``attempt`` override that cannot accept it.""" + + class _LegacyAttempt(backend_module.Backend): + def attempt(self, task, skill, memory): # no sample_id parameter + return "TOOL_CALL: search" + + response, called = _LegacyAttempt().attempt_with_tools( + _rule_task(_TOOL_CHECK), "", "", ["search"], sample_id=3 + ) + assert (response, called) == ("TOOL_CALL: search", ["search"]) + + +@pytest.mark.parametrize( + "cls", + [ + backend_module.Backend, + backend_module.MockBackend, + backend_module.PiCliBackend, + backend_module.ClaudeCliBackend, + backend_module.OpenCodeCliBackend, + backend_module.CodexCliBackend, + backend_module.CopilotCliBackend, + backend_module.CursorCliBackend, + backend_module.DualBackend, + backend_module.AzureOpenAIBackend, + backend_module.AzureResponsesBackend, + ], + ids=lambda cls: cls.__name__, +) +def test_every_shipped_tool_route_accepts_sample_id(cls) -> None: + """Signature contract only: no CLI, network, or credential is touched.""" + parameter = inspect.signature(cls.attempt_with_tools).parameters["sample_id"] + assert parameter.default == 0