fix(sleep): thread-safe backend cache + redact exports

skillopt_sleep/__main__.py

@@ -44,6 +44,7 @@
     json_safe,
     latest_staging,
     pending_staged_skills,
+    redact_secrets,
     staged_skills,
 )
 from skillopt_sleep.staging import adopt as adopt_staging
@@ -333,15 +334,16 @@ def _handoff_dir_for(cfg) -> str:
 
 
 def _redact_deep(obj):
-    """Redact secret-looking substrings in every string of a JSON-like tree."""
+    """Redact secrets key-aware across the whole structure (see redact_secrets).
+
+    This used to recurse values and only scrub string leaves, losing the
+    mapping-key context — so ``{"api_key": "x"}`` leaked. Delegating to the
+    key-aware ``redact_secrets`` walker fixes every output boundary that routes
+    through this helper (--json, digests/snapshot files, gate_trials, extra,
+    display) at once, keeping them all consistent.
+    """
     from skillopt_sleep.staging import redact_secrets
-    if isinstance(obj, str):
-        return redact_secrets(obj)
-    if isinstance(obj, list):
-        return [_redact_deep(x) for x in obj]
-    if isinstance(obj, dict):
-        return {k: _redact_deep(v) for k, v in obj.items()}
-    return obj
+    return redact_secrets(obj)
 
 
 def _display_error(exc: object) -> str:
@@ -483,7 +485,7 @@ def _handoff_mine_and_pin(cfg, args, backend, snapshot: str, dry: bool):
     # NOT marked reviewed: feeding this snapshot back through --tasks-file
     # with a real backend must still hit the human-review gate above. The
     # driver itself loads it directly, with the same trust as in-cycle mining.
-    write_tasks_file(snapshot, _redact_deep(payload))
+    write_tasks_file(snapshot, redact_secrets(payload))
     print(
         f"[sleep] handoff: pinned {len(tasks)} tasks -> {snapshot}",
         file=sys.stderr if args.json else sys.stdout,
@@ -807,9 +809,9 @@ def cmd_harvest(args) -> int:
     )
     output_path = ""
     if getattr(args, "output", ""):
-        output_path = write_tasks_file(args.output, payload)
+        output_path = write_tasks_file(args.output, redact_secrets(payload))
     if args.json:
-        json_payload = dict(payload)
+        json_payload = redact_secrets(payload)
         if output_path:
             json_payload["output"] = output_path
         print(json.dumps(json_payload, ensure_ascii=False, indent=2))

skillopt_sleep/backend.py

@@ -27,6 +27,7 @@
 import shutil
 import subprocess
 import tempfile
+import threading
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -332,6 +333,12 @@ def __init__(self, model: str = "", timeout: int = 180) -> None:
         self._cache: Dict[str, str] = {}
         self.last_call_error = ""
         self.last_reflect_raw = ""
+        # Guards _cache/_tokens against concurrent mutation on the opt-in
+        # parallel replay path (SKILLOPT_SLEEP_WORKERS>1). The model call
+        # itself stays outside the lock so parallel workers can overlap.
+        self._lock = threading.Lock()
+        # Per-thread token delta for call-local accounting under parallel replay.
+        self._thread_local = threading.local()
 
     # subclasses override --------------------------------------------------
     def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
@@ -340,16 +347,41 @@ def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
     def _cached_call(self, key: str, prompt: str, *, max_tokens: int = 1024) -> str:
         kind = key.split(":", 1)[0]
         ev = getattr(self, "evidence", None)
-        if key in self._cache:
+        with self._lock:
+            cached = self._cache.get(key)
+        if cached is not None:
+            # cover: later cache hit must not report the previous call's delta.
+            self._thread_local.delta = 0
             # cache hits log key-only (the full text is on the original miss event)
             if ev is not None:
                 ev.log("replay", "model_call", kind=kind, cache_hit=True, key=key,
                        phase=getattr(self, "evidence_phase", ""), backend=self.name,
                        model=self.model)
-            return self._cache[key]
+            return cached
+        # The model call is intentionally outside the lock so parallel workers
+        # over the same backend overlap; a concurrent miss may duplicate a call,
+        # but _cache/_tokens reads+writes below are atomic.
         out = self._call(prompt, max_tokens=max_tokens)
-        self._tokens += len(prompt) // 4 + len(out) // 4
-        self._cache[key] = out
+        delta = len(prompt) // 4 + len(out) // 4
+        with self._lock:
+            # Charge every real call's tokens. A concurrent miss may duplicate a
+            # paid call; we count each real call rather than undercount. The cache
+            # dedup below may reuse another worker's success, but the model call
+            # above still consumed tokens.
+            self._tokens += delta
+            existing = self._cache.get(key)
+            if existing:
+                # A success was cached by another worker; prefer it (dedup) so an
+                # empty/duplicate never overwrites a concurrent success.
+                out = existing
+            elif out:
+                # This worker succeeded and nothing is cached: cache it.
+                self._cache[key] = out
+            # else: empty result + nothing cached -> don't cache (transient failure)
+        # Call-local accounting: record this call's delta on the calling thread
+        # so parallel replay_one() reads its own cost, not a before/after total
+        # that another worker inflates.
+        self._thread_local.delta = delta
         if ev is not None:
             ev.log("replay", "model_call", kind=kind, cache_hit=False, key=key,
                    phase=getattr(self, "evidence_phase", ""), backend=self.name,
@@ -554,8 +586,33 @@ def _explain(c: str) -> str:
                 ))
         return edits
 
+    def _cache_get(self, key: str) -> str | None:
+        """Thread-safe cache read (route subclass cache access through this)."""
+        with self._lock:
+            return self._cache.get(key)
+
+    def _cache_pop(self, key: str) -> str | None:
+        """Thread-safe cache pop — used to drop a failed entry without racing."""
+        with self._lock:
+            return self._cache.pop(key, None)
+
+    def _cache_pop_if(self, key: str, expected: str | None) -> None:
+        """Drop a cache entry only if it still holds ``expected``.
+
+        A failed caller must not delete a successful result another worker just
+        stored for the same key, so we only pop our own (empty) value.
+        """
+        with self._lock:
+            if self._cache.get(key) == expected:
+                self._cache.pop(key, None)
+
     def tokens_used(self) -> int:
-        return self._tokens
+        with self._lock:
+            return self._tokens
+
+    def token_delta(self) -> int:
+        """Token cost of the most recent call on THIS thread (call-local)."""
+        return getattr(self._thread_local, "delta", 0)
 
 
 # ── Pi CLI backend ────────────────────────────────────────────────
@@ -604,13 +661,13 @@ def _set_call_error(self, message: object) -> None:
 
     def _cached_call(self, key: str, prompt: str, *, max_tokens: int = 1024) -> str:
         """Do not make a transient Pi failure sticky in the response cache."""
-        if key in self._cache:
+        if self._cache_get(key) is not None:
             # A cached success must not expose an unrelated previous failure
             # through diagnostics/evidence attached to this call.
             self.last_call_error = ""
         out = super()._cached_call(key, prompt, max_tokens=max_tokens)
         if not out:
-            self._cache.pop(key, None)
+            self._cache_pop_if(key, out)
         return out
 
     def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
@@ -856,7 +913,11 @@ def attempt_with_tools(self, task, skill, memory, tools):
                     "Claude CLI could not be executed: %s", exc,
                 )
                 resp = ""
-            self._tokens += len(prompt) // 4 + len(resp) // 4
+            delta = len(prompt) // 4 + len(resp) // 4
+            self._tokens += delta
+            # Call-local accounting: replay_one() reads token_delta() after
+            # attempt_with_tools(), so record this call's cost on the thread.
+            self._thread_local.delta = delta
             called: List[str] = []
             if os.path.exists(calllog):
                 with open(calllog) as f:
@@ -1239,11 +1300,11 @@ def _verify_tool_allowlist(self, env: Dict[str, str], work: str, agent: str, exp
 
     def _cached_call(self, key: str, prompt: str, *, max_tokens: int = 1024) -> str:
         """Keep failed OpenCode calls out of the cache."""
-        if key in self._cache:
+        if self._cache_get(key) is not None:
             self.last_call_error = ""
         out = super()._cached_call(key, prompt, max_tokens=max_tokens)
         if not out:
-            self._cache.pop(key, None)
+            self._cache_pop_if(key, out)
         return out
 
     def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
@@ -1356,10 +1417,14 @@ def attempt_with_tools(
                     name for name, tool_id in project.tool_mapping.items() if tool_id in called
                 ]
         except OpenCodeError as exc:
-            self._tokens += exc.prompt_chars // 4
+            delta = exc.prompt_chars // 4
+            self._tokens += delta
+            self._thread_local.delta = delta
             self.last_call_error = str(exc)
             return "", []
-        self._tokens += len(prompt) // 4 + len(text) // 4
+        delta = len(prompt) // 4 + len(text) // 4
+        self._tokens += delta
+        self._thread_local.delta = delta
         return text, called_tools
 
 
@@ -1637,7 +1702,9 @@ def attempt_with_tools(self, task, skill, memory, tools):
                 self.last_call_error = (
                     f"codex exec (tools) exited {proc.returncode}: {(proc.stderr or '')[:500]}"
                 )
-            self._tokens += len(prompt) // 4 + len(resp) // 4
+            delta = len(prompt) // 4 + len(resp) // 4
+            self._tokens += delta
+            self._thread_local.delta = delta
             called: List[str] = []
             if os.path.exists(calllog):
                 with open(calllog) as f:
@@ -1903,7 +1970,9 @@ def attempt_with_tools(self, task, skill, memory, tools):
                 resp = self._parse_jsonl_response(proc.stdout or "")
             except Exception:
                 resp = ""
-            self._tokens += len(prompt) // 4 + len(resp) // 4
+            delta = len(prompt) // 4 + len(resp) // 4
+            self._tokens += delta
+            self._thread_local.delta = delta
             called: List[str] = []
             if os.path.exists(calllog):
                 with open(calllog) as f:
@@ -2193,6 +2262,15 @@ def _call(self, prompt, *, max_tokens=1024):
     def tokens_used(self):
         return self.target.tokens_used() + self.optimizer.tokens_used()
 
+    def token_delta(self) -> int:
+        # Call-local cost for a replay attempt: replay_one() drives
+        # attempt/attempt_with_tools -> the TARGET backend, so the per-attempt
+        # delta is the target's. The optimizer only appears in replay via
+        # judge() on the rare model-judge fallback (rule/exact/answer tasks are
+        # scored locally, 0 tokens); that cost is still counted in the aggregate
+        # tokens_used() (target + optimizer), so the total is not undercounted.
+        return getattr(self.target, "token_delta", lambda: 0)()
+
 
 # ── Azure OpenAI backend (gpt-5.x via managed identity) ───────────────────────
 

skillopt_sleep/replay.py

@@ -36,13 +36,16 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
     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, sample_id=sample_id)
     latency_ms = (time.time() - t0) * 1000.0
-    tokens = max(0, backend.tokens_used() - tok_before)
+    # Call-local token accounting (thread-safe under parallel replay): use the
+    # backend's per-call delta rather than a before/after global total, which
+    # another overlapping worker would inflate.
+    token_delta = getattr(backend, "token_delta", None)
+    tokens = token_delta() if token_delta else 0
     # 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

skillopt_sleep/staging.py

@@ -1089,13 +1089,13 @@ def write_staging(
         (
             os.path.join(out, "report.json"),
             json.dumps(
-                json_safe(report.to_dict()),
+                json_safe(redact_secrets(report.to_dict())),
                 ensure_ascii=False,
                 indent=2,
                 allow_nan=False,
             ),
         ),
-        (os.path.join(out, "report.md"), report_md),
+        (os.path.join(out, "report.md"), redact_secrets(report_md)),
         # The manifest is the publication marker and must always be last.
         (
             os.path.join(out, "manifest.json"),