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/replay.py

@@ -36,16 +36,32 @@ 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()
+    # Backends that only expose the older tokens_used() contract (no per-call
+    # token_delta) need a before/after difference on the same thread to report
+    # the call-local cost; snapshot the total before the attempt for them.
+    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)
     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)
-    # 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
+    # Call-local token accounting (thread-safe under parallel replay): prefer the
+    # backend's per-call delta (CliBackend/DualBackend use a thread-local delta).
+    # That method reports exact cost, including a known zero for a cache hit, so
+    # no text-length estimate is substituted. A backend that only implements the
+    # older tokens_used() contract has no per-call delta, so use its same-thread
+    # before/after difference and fall back to a length estimate only when it
+    # reports no tracking (zero) at all.
+    if token_delta_fn is not None:
+        tokens = token_delta_fn()
+    else:
+        tokens = max(0, backend.tokens_used() - (tokens_before or 0))
+        # A backend without token_delta() that still changed tokens_used()
+        # reports that real difference above. Only an unchanged total means the
+        # backend does not track tokens (e.g. a mock), so approximate then.
+        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:

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"),

tests/test_azure_usage_accounting.py

@@ -0,0 +1,170 @@
+"""Azure provider-usage accounting regressions: single-owner charging.
+
+A backend that reports provider usage (AzureOpenAI / AzureResponses) must be
+charged exactly once in ``_cached_call`` — with the provider's own token count,
+never double-charged by the ``len//4`` length estimate, and never charged on a
+cache hit. (The maintainer's reproduction: a 30-token provider usage was being
+recorded as a 110-token length estimate because ``_call`` recorded usage and
+``_cached_call`` then recorded the length estimate on top.)
+
+Also covers the OpenCode error path routing through ``_record_delta``.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest import mock
+
+from skillopt_sleep.backend import AzureOpenAIBackend, AzureResponsesBackend, OpenCodeCliBackend
+
+
+class _ChatResp:
+    def __init__(self, text, prompt_tokens, completion_tokens):
+        self.choices = [SimpleNamespace(message=SimpleNamespace(content=text))]
+        self.usage = SimpleNamespace(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens)
+
+
+class _FakeChatClient:
+    """Scripted chat.completions.create returning a _ChatResp or raising."""
+
+    def __init__(self, replies):
+        self.replies = list(replies)
+        self.calls = []
+
+    def create(self, **kwargs):
+        self.calls.append(kwargs)
+        item = self.replies.pop(0)
+        if isinstance(item, Exception):
+            raise item
+        return item
+
+
+class _ResponsesResp:
+    def __init__(self, text, input_tokens, output_tokens):
+        self.output_text = text
+        self.usage = SimpleNamespace(input_tokens=input_tokens, output_tokens=output_tokens)
+
+
+class _FakeResponsesClient:
+    def __init__(self, replies):
+        self.replies = list(replies)
+        self.calls = []
+
+    def create(self, **kwargs):
+        self.calls.append(kwargs)
+        item = self.replies.pop(0)
+        if isinstance(item, Exception):
+            raise item
+        return item
+
+
+def _azure_chat(replies):
+    be = AzureOpenAIBackend(deployment="gpt-5.5")
+    be._client = SimpleNamespace(chat=SimpleNamespace(completions=_FakeChatClient(replies)))
+    return be
+
+
+def _azure_responses(replies):
+    be = AzureResponsesBackend(deployment="gpt-5.5", endpoints=["https://t.openai.azure.com/"])
+    fake = SimpleNamespace(responses=_FakeResponsesClient(replies))
+    be._next_endpoint = lambda: be.endpoints[0]
+    be._client_for = lambda ep: fake
+    return be
+
+
+def test_azure_chat_single_call_charges_exact_usage():
+    """A single Azure chat call must charge the provider usage, not len//4."""
+    be = _azure_chat([_ChatResp("ok", 10, 20)])
+    with mock.patch("time.sleep"):
+        out = be._cached_call("k:1", "x" * 400)
+    assert out == "ok"
+    # provider usage = 30; len//4 of 400+2 would be ~100 — must NOT be that.
+    assert be._tokens == 30, f"expected exact provider usage 30, got {be._tokens}"
+    assert be.token_delta() == 30
+
+
+def test_azure_chat_empty_retry_then_success_accumulates():
+    """Empty-response retry + success must accumulate usage across paid attempts."""
+    be = _azure_chat([_ChatResp("", 7, 0), _ChatResp("ok", 10, 20)])
+    with mock.patch("time.sleep"):
+        out = be._cached_call("k:1", "hello")
+    assert out == "ok"
+    assert be._tokens == 7 + 30, f"expected accumulated 37, got {be._tokens}"
+    assert be.token_delta() == 37
+
+
+def test_azure_chat_cache_hit_does_not_charge():
+    """A cache hit must reset the call-local delta and leave the aggregate alone."""
+    be = _azure_chat([_ChatResp("ok", 10, 20)])
+    with mock.patch("time.sleep"):
+        be._cached_call("k:1", "hello")
+    assert be._tokens == 30
+    assert be.token_delta() == 30
+    with mock.patch("time.sleep"):
+        out2 = be._cached_call("k:1", "hello")
+    assert out2 == "ok"
+    assert be._tokens == 30, "cache hit changed the aggregate"
+    assert be.token_delta() == 0, "cache hit leaked the prior delta"
+
+
+def test_azure_responses_single_call_charges_exact_usage():
+    be = _azure_responses([_ResponsesResp("ok", 12, 18)])
+    with mock.patch("time.sleep"):
+        out = be._cached_call("k:1", "x" * 400)
+    assert out == "ok"
+    assert be._tokens == 30, f"expected exact provider usage 30, got {be._tokens}"
+    assert be.token_delta() == 30
+
+
+def test_azure_responses_empty_retry_then_success_accumulates():
+    be = _azure_responses([_ResponsesResp("", 5, 0), _ResponsesResp("ok", 12, 18)])
+    with mock.patch("time.sleep"):
+        out = be._cached_call("k:1", "hello")
+    assert out == "ok"
+    assert be._tokens == 5 + 30, f"expected accumulated 35, got {be._tokens}"
+    assert be.token_delta() == 35
+
+
+def test_azure_responses_cache_hit_does_not_charge():
+    be = _azure_responses([_ResponsesResp("ok", 12, 18)])
+    with mock.patch("time.sleep"):
+        be._cached_call("k:1", "hello")
+    assert be._tokens == 30
+    with mock.patch("time.sleep"):
+        be._cached_call("k:1", "hello")
+    assert be._tokens == 30
+    assert be.token_delta() == 0
+
+
+def test_opencode_error_path_uses_record_delta(monkeypatch):
+    """The OpenCode error path must route prompt-only cost through _record_delta."""
+    import contextlib
+    from types import SimpleNamespace
+
+    import skillopt_sleep.backend as bm
+    from skillopt_sleep.backend import OpenCodeCliBackend
+
+    b = OpenCodeCliBackend(model="", opencode_path="opencode", tool_replay=True)
+    monkeypatch.setattr(bm, "_opencode_temporary_workspace", lambda *a, **k: contextlib.nullcontext())
+
+    def _fail(*args, **kwargs):
+        raise bm.OpenCodeError("boom", prompt_chars=100)
+
+    monkeypatch.setattr(bm, "_prepare_opencode_replay_project", _fail)
+
+    task = SimpleNamespace(intent="intent", context_excerpt="ctx")
+    out, called = b.attempt_with_tools(task, skill="s", memory="m", tools=["search"])
+
+    assert out == "" and called == []
+    assert b.token_delta() == 100 // 4, "OpenCode error path did not route through _record_delta"
+    assert b._tokens == 100 // 4, f"expected 25, got {b._tokens}"
+
+
+def test_azure_chat_paid_empty_then_terminal_error_keeps_usage():
+    """A paid empty response followed by exhausted exception retries must keep the
+    exact provider usage (7), not fall back to the len//4 length estimate."""
+    be = _azure_chat([_ChatResp("", 7, 0)] + [RuntimeError("boom")] * 4)
+    with mock.patch("time.sleep"):
+        out = be._cached_call("k:1", "x" * 400)
+    assert out == ""
+    assert be._tokens == 7, f"expected paid usage 7, got {be._tokens}"
+    assert be.token_delta() == 7, f"expected call-local delta 7, got {be.token_delta()}"