fix(sleep): handoff no-task cleanup + corrupt tasks.json fallback

Follow-up to #125. Two resumability defects in the handoff driver:

1. A no-task night left digests.json (often []) pinned, so every later run
   reloaded the empty pin and never called harvest_for_config again. Remove the
   pin on terminal no-task completion so the next run re-harvests.

2. An interrupted/corrupt tasks.json snapshot raised an uncaught
   JSONDecodeError in _run_handoff. Wrap load_tasks_file in try/except and
   re-mine instead of crashing.

Adds regression tests for both.

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

.gitignore

@@ -24,6 +24,9 @@ logs/
 external/
 # SkillOpt-Sleep runtime state (staging proposals, config, diagnostics, cron logs)
 .skillopt-sleep/
+# SkillOpt-Sleep handoff-backend round data (prompts/answers derived from transcripts)
+.skillopt-sleep-handoff/
+.skillopt-sleep-handoff.night*.done/
 
 /BabyVision/
 /MMRB/

CHANGELOG.md

@@ -4,6 +4,19 @@ All notable changes to SkillOpt are documented here. This project adheres to
 [Semantic Versioning](https://semver.org/) and the format is based on
 [Keep a Changelog](https://keepachangelog.com/).
 
+## [Unreleased]
+
+### Added
+- **Handoff backend** (`--backend handoff`) for SkillOpt-Sleep — runs the
+  sleep cycle with no model subprocess or API key: the engine writes each
+  pending model call to `PROMPTS.md`/`pending.json` (exit code 3) and the
+  user's own agent session answers into `answers/<id>.md`; re-running the
+  same command resumes statelessly from the answers (typically 3–6 rounds
+  per night). Mined tasks are pinned per night so answering sessions cannot
+  shift the task set. Ships a `/skillopt-sleep-handoff` Claude Code command
+  that automates the loop with fresh-context subagents to protect the
+  held-out gate.
+
 ## [0.2.0] — 2026-07-02
 
 The headline of this release is **SkillOpt-Sleep**: a nightly offline

plugins/README.md

@@ -22,7 +22,7 @@ sleep** idea (short-term experience → long-term competence).
 
 | Platform | Folder | Mechanism | Status |
 |---|---|---|---|
-| **Claude Code** | [`claude-code/`](claude-code) | `.claude-plugin` + `/skillopt-sleep` command + skill + hooks | full, installable |
+| **Claude Code** | [`claude-code/`](claude-code) | `.claude-plugin` + `/skillopt-sleep` + `/skillopt-sleep-handoff` commands + skill + hooks | full, installable |
 | **Codex** | [`codex/`](codex) | user-level `skillopt-sleep` skill + shared runner | full |
 | **Copilot** | [`copilot/`](copilot) | MCP server (`sleep_*` tools) + `copilot-instructions` | full (MCP) |
 | **Devin** | [`devin/`](devin) | MCP server (`sleep_*` tools) + Devin ATIF-v1.7 harvest + `.devin/rules` | full (MCP) |
@@ -149,6 +149,47 @@ The reward can weight not just correctness but **cost and speed**, so a skill ca
 learn to be cheaper and faster, not only more accurate. *What it does for you:*
 "answer directly instead of opening five files" becomes a learned habit.
 
+### `--backend handoff` — session-executed calls (no API subprocess)
+
+For subscription seats and environments where the engine shouldn't spawn
+`claude -p` / API calls itself. The engine still runs every deterministic
+stage (harvest → mine → replay scoring → gate → stage), but each model call
+(attempt / judge / reflect) is written to a prompt file that **your own agent
+session answers between engine runs**:
+
+```bash
+python -m skillopt_sleep run --backend handoff --project "$(pwd)"
+# exit 3 => .skillopt-sleep-handoff/PROMPTS.md + pending.json were written
+#   answer each prompt (each in a FRESH context) into answers/<id>.md
+# re-run the same command => it resumes from the answers and either
+#   finishes (exit 0) or stages the next prompt batch (exit 3)
+```
+
+A typical night converges in 3–6 rounds: baseline attempts → reflect →
+candidate re-scoring per accepted edit. Resume is stateless — replay is
+deterministic and answers are cached by prompt hash, so re-running skips
+everything already answered. Mined tasks are pinned to
+`.skillopt-sleep-handoff/tasks.json` on the first round, so the sessions that
+answer the prompts can't shift the task set and invalidate earlier answers.
+On a completed real run the handoff directory is archived to
+`.skillopt-sleep-handoff.night<N>.done`.
+
+On Claude Code, `/skillopt-sleep-handoff run` drives the whole loop for you,
+answering each prompt in an isolated fresh-context subagent.
+
+**Integrity rule:** answer every prompt in a fresh context (a subagent with no
+conversation history). Answering from a session that has already seen the
+mined tasks and their references contaminates the held-out gate and fakes the
+improvement score.
+
+*What it does for you:* the sleep cycle runs entirely on your interactive
+session's subscription budget — no API key, no headless subprocess — while the
+gate, splits, and staging discipline stay in the engine.
+
+Limitations: `--rollouts-k > 1` gives no contrastive spread (identical prompt
+→ identical answer file), and tool-loop tasks fall back to the single-shot
+`TOOL_CALL:` marker convention.
+
 ### `schedule` / `unschedule` — set it and forget it
 
 Built-in nightly scheduling (no manual cron):
@@ -176,7 +217,7 @@ schedule, if you trust it).
 
 | Flag | Default | Meaning |
 |---|---|---|
-| `--backend mock\|claude\|codex\|copilot` | `mock` | who runs/optimizes (mock = free) |
+| `--backend mock\|claude\|codex\|copilot\|handoff` | `mock` | who runs/optimizes (mock = free; handoff = your own session answers) |
 | `--preferences "..."` | – | your house rules, as a prior |
 | `--gate on\|off` | `on` | strict held-out gate vs. greedy |
 | `--rollouts-k K` | `1` | multi-rollout contrastive reflection |

plugins/claude-code/README.md

@@ -60,6 +60,9 @@ they shell out to the CLIs you already have.
 /skillopt-sleep run         # full cycle: stages a reviewed proposal (still no live edits)
 /skillopt-sleep status      # see history + the latest staged proposal
 /skillopt-sleep adopt       # apply the staged proposal to CLAUDE.md / SKILL.md (with backup)
+
+/skillopt-sleep-handoff run # same cycle, but THIS session answers the model calls
+                            # (no claude -p subprocess, no API key — subscription-friendly)
 ```
 
 Or call the engine directly (Python ≥ 3.10):
@@ -74,6 +77,26 @@ Default backend is **`mock`** — deterministic, no API spend — so you can try
 plumbing for free. Switch to `--backend claude` or `--backend codex` for genuine
 improvement on your own budget.
 
+### Handoff mode (session answers the model calls)
+
+`--backend handoff` runs the cycle without any model subprocess: the engine
+executes the deterministic stages and writes every model call it needs to
+`.skillopt-sleep-handoff/PROMPTS.md` + `pending.json` (exit code 3). You (or
+the `/skillopt-sleep-handoff` command, which automates the loop with isolated
+fresh-context subagents) write each raw answer to `answers/<id>.md` and re-run
+the same command; it resumes from the answers and either finishes or stages
+the next batch. Typically 3–6 rounds per night.
+
+```bash
+python -m skillopt_sleep run --backend handoff --project "$(pwd)"
+# ... answer .skillopt-sleep-handoff/PROMPTS.md into answers/<id>.md ...
+python -m skillopt_sleep run --backend handoff --project "$(pwd)"   # resume
+```
+
+Answer every prompt in a **fresh context** — a session that has already seen
+the mined tasks and their references would contaminate the held-out gate.
+Details: [the plugins README](../README.md#--backend-handoff--session-executed-calls-no-api-subprocess).
+
 ## Does it actually improve? (real models, public benchmark)
 
 SkillOpt-Sleep is validated against [gbrain-evals](https://github.com/garrytan/gbrain-evals)'

plugins/claude-code/commands/skillopt-sleep-handoff.md

@@ -0,0 +1,67 @@
+---
+description: Run the SkillOpt-Sleep cycle with the handoff backend — no API subprocess; this session answers the engine's model calls via prompt/answer files, in isolated fresh-context subagents
+argument-hint: "[run | dry-run] [--preferences \"...\"] (default: run)"
+allowed-tools: Bash, Read, Write, Task
+---
+
+# /skillopt-sleep-handoff — session-executed sleep cycle
+
+You are driving **SkillOpt-Sleep in handoff mode**: the Python engine runs
+every deterministic stage (harvest → mine → replay scoring → gate → stage)
+and outsources each model call (attempt / judge / reflect) to YOU via
+prompt files. No `claude -p` subprocess, no API key — the model work runs
+on this session's budget, but each prompt MUST be answered in a fresh,
+isolated context so the validation gate stays honest.
+
+## Requested action: $ARGUMENTS
+
+(If `$ARGUMENTS` is empty, treat it as `run`.)
+
+## The loop
+
+Repeat until the engine exits 0 (done) — at most 8 rounds:
+
+1. **Run the engine** via the bundled runner:
+
+   ```bash
+   "${CLAUDE_PLUGIN_ROOT}/scripts/sleep.sh" <action> --backend handoff --project "$(pwd)" --scope invoked
+   ```
+
+   - exit 0 → the night is complete; go to "Finish" below.
+   - exit 3 → pending model calls; continue with step 2.
+   - anything else → stop and show the user the error output.
+
+2. **Read the batch**: `Read` `.skillopt-sleep-handoff/pending.json` in the
+   project. Each entry has `id`, `prompt`, `max_tokens`, `answer_file`.
+
+3. **Answer each prompt in ISOLATION** — this is the integrity rule:
+   - For each entry, launch a subagent (Task tool) whose ENTIRE input is
+     the `prompt` text verbatim. Add nothing: no summary of this session,
+     no mention of SkillOpt, no other prompts from the batch.
+   - Take the subagent's reply and `Write` the raw answer text (no
+     commentary, no code fences) to the entry's `answer_file`.
+   - NEVER answer from this session's own context — you have seen the
+     mined tasks and their references, so inline answers would contaminate
+     the held-out gate and fake the improvement score.
+
+4. **Re-run the same engine command** — it resumes from the answers
+   directory and either finishes or stages the next batch.
+
+## Finish
+
+- `Read` the `report.md` in the staging dir the engine printed and show
+  the user: held-out baseline → candidate score, the gate decision, the
+  proposed edits, and where the proposal is staged.
+- Tell the user nothing live changed; offer `/skillopt-sleep adopt`.
+- The engine archives `.skillopt-sleep-handoff/` on a completed real run;
+  do not delete it yourself.
+
+## Safety reminders
+
+- **Never** edit `CLAUDE.md` or `SKILL.md` yourself — only `adopt` does
+  that, with a backup.
+- Mined tasks are pinned to `.skillopt-sleep-handoff/tasks.json` on round
+  one, so sessions created while answering prompts cannot shift the task
+  set. Do not edit that file.
+- If a batch looks like it contains secrets or content the user would not
+  want re-processed, stop and ask before answering.

skillopt/config.py

@@ -125,6 +125,9 @@ _FLATTEN_MAP: dict[str, str] = {
     "evaluation.use_gate": "use_gate",
     "evaluation.gate_metric": "gate_metric",
     "evaluation.gate_mixed_weight": "gate_mixed_weight",
+    "evaluation.use_semantic_density": "use_semantic_density",
+    "evaluation.semantic_density_weight": "semantic_density_weight",
+    "evaluation.leading_words": "leading_words",
     "evaluation.sel_env_num": "sel_env_num",
     "evaluation.test_env_num": "test_env_num",
     "evaluation.eval_test": "eval_test",
@@ -158,7 +161,7 @@ def _load_yaml(path: str, _visited: set[str] | None = None) -> dict:
         raise ValueError(f"Circular _base_ inheritance: {abs_path}")
     _visited.add(abs_path)
 
-    with open(abs_path) as f:
+    with open(abs_path, encoding="utf-8") as f:
         cfg = yaml.safe_load(f) or {}
 
     base_ref = cfg.pop("_base_", None)

skillopt/engine/trainer.py

@@ -709,26 +709,26 @@ class ReflACTTrainer:
             effort=cfg.get("claude_code_exec_effort", cfg.get("reasoning_effort", "medium")),
             max_thinking_tokens=cfg.get("claude_code_exec_max_thinking_tokens", 16384),
         )
-        configure_qwen_chat(
-            base_url=cfg.get("qwen_chat_base_url") or None,
-            api_key=cfg.get("qwen_chat_api_key") or None,
-            temperature=cfg.get("qwen_chat_temperature"),
-            timeout_seconds=cfg.get("qwen_chat_timeout_seconds"),
-            max_tokens=cfg.get("qwen_chat_max_tokens"),
-            enable_thinking=cfg.get("qwen_chat_enable_thinking"),
-            optimizer_base_url=cfg.get("optimizer_qwen_chat_base_url") or None,
-            optimizer_api_key=cfg.get("optimizer_qwen_chat_api_key") or None,
-            optimizer_temperature=cfg.get("optimizer_qwen_chat_temperature"),
-            optimizer_timeout_seconds=cfg.get("optimizer_qwen_chat_timeout_seconds"),
-            optimizer_max_tokens=cfg.get("optimizer_qwen_chat_max_tokens"),
-            optimizer_enable_thinking=cfg.get("optimizer_qwen_chat_enable_thinking"),
-            target_base_url=cfg.get("target_qwen_chat_base_url") or None,
-            target_api_key=cfg.get("target_qwen_chat_api_key") or None,
-            target_temperature=cfg.get("target_qwen_chat_temperature"),
-            target_timeout_seconds=cfg.get("target_qwen_chat_timeout_seconds"),
-            target_max_tokens=cfg.get("target_qwen_chat_max_tokens"),
-            target_enable_thinking=cfg.get("target_qwen_chat_enable_thinking"),
-        )
+        configure_qwen_chat(
+            base_url=cfg.get("qwen_chat_base_url") or None,
+            api_key=cfg.get("qwen_chat_api_key") or None,
+            temperature=cfg.get("qwen_chat_temperature"),
+            timeout_seconds=cfg.get("qwen_chat_timeout_seconds"),
+            max_tokens=cfg.get("qwen_chat_max_tokens"),
+            enable_thinking=cfg.get("qwen_chat_enable_thinking"),
+            optimizer_base_url=cfg.get("optimizer_qwen_chat_base_url") or None,
+            optimizer_api_key=cfg.get("optimizer_qwen_chat_api_key") or None,
+            optimizer_temperature=cfg.get("optimizer_qwen_chat_temperature"),
+            optimizer_timeout_seconds=cfg.get("optimizer_qwen_chat_timeout_seconds"),
+            optimizer_max_tokens=cfg.get("optimizer_qwen_chat_max_tokens"),
+            optimizer_enable_thinking=cfg.get("optimizer_qwen_chat_enable_thinking"),
+            target_base_url=cfg.get("target_qwen_chat_base_url") or None,
+            target_api_key=cfg.get("target_qwen_chat_api_key") or None,
+            target_temperature=cfg.get("target_qwen_chat_temperature"),
+            target_timeout_seconds=cfg.get("target_qwen_chat_timeout_seconds"),
+            target_max_tokens=cfg.get("target_qwen_chat_max_tokens"),
+            target_enable_thinking=cfg.get("target_qwen_chat_enable_thinking"),
+        )
         configure_minimax_chat(
             base_url=cfg.get("minimax_base_url") or None,
             api_key=cfg.get("minimax_api_key") or None,
@@ -964,6 +964,15 @@ class ReflACTTrainer:
                 f"got {gate_metric!r}"
             )
         gate_mixed_weight = float(cfg.get("gate_mixed_weight", 0.5))
+        use_semantic_density = bool(cfg.get("use_semantic_density", False))
+        semantic_density_weight = float(cfg.get("semantic_density_weight", 0.05))
+        leading_words_raw = cfg.get("leading_words", None)
+        leading_words = None
+        if leading_words_raw is not None:
+            if isinstance(leading_words_raw, str):
+                leading_words = [w.strip() for w in leading_words_raw.split(",") if w.strip()]
+            else:
+                leading_words = list(leading_words_raw)
         if not 0.0 <= gate_mixed_weight <= 1.0:
             raise ValueError(
                 f"evaluation.gate_mixed_weight must be in [0, 1], "
@@ -1003,6 +1012,10 @@ class ReflACTTrainer:
             baseline_hard, baseline_soft = compute_score(baseline_results)
             current_score = select_gate_score(
                 baseline_hard, baseline_soft, gate_metric, gate_mixed_weight,
+                skill_content=skill_init,
+                use_semantic_density=use_semantic_density,
+                semantic_density_weight=semantic_density_weight,
+                leading_words=leading_words,
             )
             best_score = current_score
             sh = skill_hash(skill_init)
@@ -1450,9 +1463,16 @@ class ReflACTTrainer:
                     cand_soft=cand_soft,
                     metric=gate_metric,
                     mixed_weight=gate_mixed_weight,
+                    use_semantic_density=use_semantic_density,
+                    semantic_density_weight=semantic_density_weight,
+                    leading_words=leading_words,
                 ) if use_gate else None
                 cand_gate_score = select_gate_score(
                     cand_hard, cand_soft, gate_metric, gate_mixed_weight,
+                    skill_content=candidate_skill,
+                    use_semantic_density=use_semantic_density,
+                    semantic_density_weight=semantic_density_weight,
+                    leading_words=leading_words,
                 )
                 if not use_gate:
                     # Validation ran (scores recorded above) but the gate is
@@ -1844,6 +1864,9 @@ class ReflACTTrainer:
                                 cand_soft=slow_sel_soft,
                                 metric=gate_metric,
                                 mixed_weight=gate_mixed_weight,
+                                use_semantic_density=use_semantic_density,
+                                semantic_density_weight=semantic_density_weight,
+                                leading_words=leading_words,
                             )
                             slow_result["selection_hard"] = slow_sel_hard
                             slow_result["selection_soft"] = slow_sel_soft
@@ -2093,6 +2116,10 @@ class ReflACTTrainer:
                     final_gate_score = select_gate_score(
                         final_selection_hard, final_selection_soft,
                         gate_metric, gate_mixed_weight,
+                        skill_content=current_skill,
+                        use_semantic_density=use_semantic_density,
+                        semantic_density_weight=semantic_density_weight,
+                        leading_words=leading_words,
                     )
                     print(
                         f"\n  [final skill val] items={fval_n} "

skillopt/evaluation/gate.py

@@ -43,11 +43,55 @@ class GateResult:
     best_step: int
 
 
+def compute_semantic_density(
+    skill_content: str,
+    leading_words: list[str] | None = None,
+) -> float:
+    """Compute the semantic density of leading words in a skill document."""
+    if not skill_content or not skill_content.strip():
+        return 0.0
+    if leading_words is None:
+        leading_words = [
+            "MUST", "ALWAYS", "NEVER", "ONLY", "CRITICAL", "IMPORTANT",
+            "RESOLVE", "PREFER", "ENSURE", "STRICT", "VERIFY"
+        ]
+    
+    # Strip metadata comments to focus purely on instruction text
+    skill = skill_content
+    for start, end in [
+        ("<!-- SLOW_UPDATE_START -->", "<!-- SLOW_UPDATE_END -->"),
+        ("<!-- APPENDIX_START -->", "<!-- APPENDIX_END -->")
+    ]:
+        while True:
+            s_idx = skill.find(start)
+            if s_idx == -1:
+                break
+            e_idx = skill.find(end, s_idx)
+            if e_idx == -1:
+                skill = skill[:s_idx] + skill[s_idx + len(start):]
+                break
+            skill = skill[:s_idx] + skill[e_idx + len(end):]
+
+    import re
+    words = re.findall(r'[a-zA-Z0-9]+', skill.lower())
+    if not words:
+        return 0.0
+    
+    leading_set = {w.lower() for w in leading_words}
+    leading_count = sum(1 for w in words if w in leading_set)
+    return leading_count / len(words)
+
+
 def select_gate_score(
     hard: float,
     soft: float,
     metric: GateMetric = "hard",
     mixed_weight: float = 0.5,
+    *,
+    skill_content: str = "",
+    use_semantic_density: bool = False,
+    semantic_density_weight: float = 0.05,
+    leading_words: list[str] | None = None,
 ) -> float:
     """Project (hard, soft) onto a single comparison metric.
 
@@ -60,17 +104,32 @@ def select_gate_score(
     mixed_weight
         For ``"mixed"``: weight given to ``soft``. Must be in ``[0, 1]``.
         Ignored for ``"hard"`` / ``"soft"``.
+    skill_content
+        The raw skill document content.
+    use_semantic_density
+        Whether to adjust the score based on semantic density of leading words.
+    semantic_density_weight
+        Scaling weight for the semantic density bonus.
+    leading_words
+        Optional custom list of high-influence words to prioritize.
     """
     if metric == "hard":
-        return float(hard)
-    if metric == "soft":
-        return float(soft)
-    if metric == "mixed":
+        score = float(hard)
+    elif metric == "soft":
+        score = float(soft)
+    elif metric == "mixed":
         w = max(0.0, min(1.0, float(mixed_weight)))
-        return (1.0 - w) * float(hard) + w * float(soft)
-    raise ValueError(
-        f"unknown gate metric {metric!r}; expected 'hard', 'soft', or 'mixed'"
-    )
+        score = (1.0 - w) * float(hard) + w * float(soft)
+    else:
+        raise ValueError(
+            f"unknown gate metric {metric!r}; expected 'hard', 'soft', or 'mixed'"
+        )
+
+    if use_semantic_density:
+        density = compute_semantic_density(skill_content, leading_words)
+        score += float(semantic_density_weight) * density
+
+    return score
 
 
 def evaluate_gate(
@@ -86,6 +145,9 @@ def evaluate_gate(
     cand_soft: float = 0.0,
     metric: GateMetric = "hard",
     mixed_weight: float = 0.5,
+    use_semantic_density: bool = False,
+    semantic_density_weight: float = 0.05,
+    leading_words: list[str] | None = None,
 ) -> GateResult:
     """Pure gate decision: compare candidate score to current/best.
 
@@ -111,6 +173,12 @@ def evaluate_gate(
         the original gate behavior.
     mixed_weight
         Weight on ``soft`` when ``metric == "mixed"``.
+    use_semantic_density
+        Whether to adjust the score based on semantic density of leading words.
+    semantic_density_weight
+        Scaling weight for the semantic density bonus.
+    leading_words
+        Optional custom list of high-influence words to prioritize.
 
     Returns
     -------
@@ -118,7 +186,16 @@ def evaluate_gate(
         Updated state; the caller decides what to do with it (print,
         mutate trainer state, log, etc.).
     """
-    cand_score = select_gate_score(cand_hard, cand_soft, metric, mixed_weight)
+    cand_score = select_gate_score(
+        cand_hard,
+        cand_soft,
+        metric,
+        mixed_weight,
+        skill_content=candidate_skill,
+        use_semantic_density=use_semantic_density,
+        semantic_density_weight=semantic_density_weight,
+        leading_words=leading_words,
+    )
 
     if cand_score > current_score:
         if cand_score > best_score:

skillopt/model/__init__.py

@@ -13,13 +13,13 @@ from skillopt.model.backend_config import (  # noqa: F401
     configure_codex_exec,
     get_claude_code_exec_config,
     get_codex_exec_config,
-    get_target_backend,
     get_optimizer_backend,
+    get_target_backend,
+    is_optimizer_chat_backend,
     is_target_chat_backend,
     is_target_exec_backend,
-    is_optimizer_chat_backend,
-    set_target_backend,
     set_optimizer_backend,
+    set_target_backend,
 )
 
 
@@ -105,6 +105,16 @@ def chat_optimizer(
             reasoning_effort=reasoning_effort,
             timeout=timeout,
         )
+    if get_optimizer_backend() == "minimax_chat":
+        return _minimax.chat_optimizer(
+            system=system,
+            user=user,
+            max_completion_tokens=max_completion_tokens,
+            retries=retries,
+            stage=stage,
+            reasoning_effort=reasoning_effort,
+            timeout=timeout,
+        )
     return _openai.chat_optimizer(
         system=system,
         user=user,
@@ -204,6 +214,18 @@ def chat_optimizer_messages(
             return_message=return_message,
             timeout=timeout,
         )
+    if get_optimizer_backend() == "minimax_chat":
+        return _minimax.chat_target_messages(
+            messages=messages,
+            max_completion_tokens=max_completion_tokens,
+            retries=retries,
+            stage=stage,
+            reasoning_effort=reasoning_effort,
+            tools=tools,
+            tool_choice=tool_choice,
+            return_message=return_message,
+            timeout=timeout,
+        )
     return _openai.chat_optimizer_messages(
         messages=messages,
         max_completion_tokens=max_completion_tokens,
@@ -440,18 +462,21 @@ def configure_qwen_chat(
     timeout_seconds: float | str | None = None,
     max_tokens: int | str | None = None,
     enable_thinking: bool | str | None = None,
+    use_max_completion_tokens: bool | str | None = None,
     optimizer_base_url: str | None = None,
     optimizer_api_key: str | None = None,
     optimizer_temperature: float | str | None = None,
     optimizer_timeout_seconds: float | str | None = None,
     optimizer_max_tokens: int | str | None = None,
     optimizer_enable_thinking: bool | str | None = None,
+    optimizer_use_max_completion_tokens: bool | str | None = None,
     target_base_url: str | None = None,
     target_api_key: str | None = None,
     target_temperature: float | str | None = None,
     target_timeout_seconds: float | str | None = None,
     target_max_tokens: int | str | None = None,
     target_enable_thinking: bool | str | None = None,
+    target_use_max_completion_tokens: bool | str | None = None,
 ) -> None:
     _qwen.configure_qwen_chat(
         base_url=base_url,
@@ -460,18 +485,21 @@ def configure_qwen_chat(
         timeout_seconds=timeout_seconds,
         max_tokens=max_tokens,
         enable_thinking=enable_thinking,
+        use_max_completion_tokens=use_max_completion_tokens,
         optimizer_base_url=optimizer_base_url,
         optimizer_api_key=optimizer_api_key,
         optimizer_temperature=optimizer_temperature,
         optimizer_timeout_seconds=optimizer_timeout_seconds,
         optimizer_max_tokens=optimizer_max_tokens,
         optimizer_enable_thinking=optimizer_enable_thinking,
+        optimizer_use_max_completion_tokens=optimizer_use_max_completion_tokens,
         target_base_url=target_base_url,
         target_api_key=target_api_key,
         target_temperature=target_temperature,
         target_timeout_seconds=target_timeout_seconds,
         target_max_tokens=target_max_tokens,
         target_enable_thinking=target_enable_thinking,
+        target_use_max_completion_tokens=target_use_max_completion_tokens,
     )
 
 

skillopt/model/minimax_backend.py

@@ -222,7 +222,7 @@ def chat_target(
     stage: str = "target",
     reasoning_effort: str | None = None,
     timeout: float | None = None,
-) -> tuple[str, dict[str, int]]:
+) -> tuple[str, dict[int]]:
     del reasoning_effort
     messages = [{"role": "system", "content": system}, {"role": "user", "content": user}]
     return _chat_messages_impl(
@@ -234,6 +234,32 @@ def chat_target(
     )
 
 
+def chat_optimizer(
+    system: str,
+    user: str,
+    max_completion_tokens: int = 16384,
+    retries: int = 5,
+    stage: str = "optimizer",
+    reasoning_effort: str | None = None,
+    timeout: float | None = None,
+) -> tuple[str, dict[int]]:
+    """Optimizer chat call. Backend stores the trained skill; uses the same
+    MiniMax-proxied OpenAI-compat endpoint as `chat_target`. Added in the
+    parallel-training fix; previously missing in skillopt 0.2.0's
+    miniamax backend, which forced the dispatcher into _openai.chat_optimizer
+    (Azure) and produced "[skip] no usable patches" for any user running
+    optimizer+target on `minimax_chat`.
+    """
+    messages = [{"role": "system", "content": system}, {"role": "user", "content": user}]
+    return _chat_messages_impl(
+        messages,
+        max_completion_tokens,
+        retries,
+        stage,
+        timeout=timeout,
+    )
+
+
 def chat_target_messages(
     messages: list[dict[str, Any]],
     max_completion_tokens: int = 16384,

skillopt/model/qwen_backend.py

@@ -1,13 +1,14 @@
 """OpenAI-compatible Qwen chat backend for optimizer and target paths."""
+
 from __future__ import annotations
 
-from dataclasses import dataclass
 import json
 import os
 import threading
 import time
 import urllib.error
 import urllib.request
+from dataclasses import dataclass
 from typing import Any
 
 from skillopt.model.common import (
@@ -28,6 +29,7 @@ class QwenChatConfig:
     temperature: float | None
     enable_thinking: bool
     deployment: str
+    use_max_completion_tokens: bool = False
 
 
 def _parse_bool(value: Any, default: bool = False) -> bool:
@@ -56,6 +58,28 @@ def _role_env(role: str, key: str, default: str) -> str:
     return os.environ.get(role_key) or os.environ.get(generic_key) or default
 
 
+# Sentinels that mean "omit this optional parameter from the request payload".
+# Reasoning models (e.g. GPT-5.x, Claude Opus 4.8) reject an explicit
+# `temperature`, so allow it to be turned off via an empty string / none / off.
+_OMIT_SENTINELS = {"", "none", "off", "null"}
+
+
+def _resolve_temperature(role: str) -> float | None:
+    """Return the temperature, or None to omit it entirely.
+
+    Unlike ``_role_env`` an *explicitly set* empty (or ``none``/``off``) value is
+    honored as "omit" instead of collapsing to the default. Precedence:
+    role-specific env -> generic env -> 0.7 default.
+    """
+    for key in (f"{role.upper()}_QWEN_CHAT_TEMPERATURE", "QWEN_CHAT_TEMPERATURE"):
+        if key in os.environ:
+            raw = os.environ[key].strip()
+            if raw.lower() in _OMIT_SENTINELS:
+                return None
+            return float(raw)
+    return 0.7
+
+
 def _initial_config(role: str) -> QwenChatConfig:
     role_upper = role.upper()
     deployment_env = "OPTIMIZER_DEPLOYMENT" if role == "optimizer" else "TARGET_DEPLOYMENT"
@@ -64,8 +88,9 @@ def _initial_config(role: str) -> QwenChatConfig:
         api_key=_role_env(role, "API_KEY", ""),
         timeout_seconds=float(_role_env(role, "TIMEOUT_SECONDS", "300") or 300),
         max_tokens=_parse_int(_role_env(role, "MAX_TOKENS", "8000"), 8000),
-        temperature=_parse_optional_float(_role_env(role, "TEMPERATURE", "0.7")),
+        temperature=_resolve_temperature(role),
         enable_thinking=_parse_bool(_role_env(role, "ENABLE_THINKING", "false")),
+        use_max_completion_tokens=_parse_bool(_role_env(role, "USE_MAX_COMPLETION_TOKENS", "false")),
         deployment=(
             os.environ.get(f"{role_upper}_QWEN_CHAT_MODEL")
             or os.environ.get("QWEN_CHAT_MODEL")
@@ -186,10 +211,14 @@ def _chat_messages_impl(
     timeout: float | None = None,
 ) -> tuple[Any, dict[str, int]]:
     config = OPTIMIZER_CONFIG if role == "optimizer" else TARGET_CONFIG
+    token_limit = min(max_completion_tokens, config.max_tokens)
+    # Reasoning models on some gateways (GPT-5.x, o-series) require
+    # `max_completion_tokens` and reject the legacy `max_tokens`.
+    token_key = "max_completion_tokens" if config.use_max_completion_tokens else "max_tokens"
     payload: dict[str, Any] = {
         "model": deployment or config.deployment,
         "messages": _json_safe(messages),
-        "max_tokens": min(max_completion_tokens, config.max_tokens),
+        token_key: token_limit,
     }
     if config.enable_thinking:
         payload["chat_template_kwargs"] = {"enable_thinking": True}
@@ -219,7 +248,7 @@ def _chat_messages_impl(
             return text, usage_info
         except Exception as e:  # noqa: BLE001
             last_err = e
-            time.sleep(min(2 ** attempt, 30))
+            time.sleep(min(2**attempt, 30))
     raise RuntimeError(f"Qwen chat call failed after {retries} retries: {last_err}")
 
 
@@ -231,18 +260,21 @@ def configure_qwen_chat(
     timeout_seconds: float | str | None = None,
     max_tokens: int | str | None = None,
     enable_thinking: bool | str | None = None,
+    use_max_completion_tokens: bool | str | None = None,
     optimizer_base_url: str | None = None,
     optimizer_api_key: str | None = None,
     optimizer_temperature: float | str | None = None,
     optimizer_timeout_seconds: float | str | None = None,
     optimizer_max_tokens: int | str | None = None,
     optimizer_enable_thinking: bool | str | None = None,
+    optimizer_use_max_completion_tokens: bool | str | None = None,
     target_base_url: str | None = None,
     target_api_key: str | None = None,
     target_temperature: float | str | None = None,
     target_timeout_seconds: float | str | None = None,
     target_max_tokens: int | str | None = None,
     target_enable_thinking: bool | str | None = None,
+    target_use_max_completion_tokens: bool | str | None = None,
 ) -> None:
     with _config_lock:
         if base_url is not None:
@@ -256,29 +288,24 @@ def configure_qwen_chat(
         if max_tokens is not None:
             os.environ["QWEN_CHAT_MAX_TOKENS"] = str(max_tokens)
         if enable_thinking is not None:
-            os.environ["QWEN_CHAT_ENABLE_THINKING"] = (
-                "true" if _parse_bool(enable_thinking) else "false"
+            os.environ["QWEN_CHAT_ENABLE_THINKING"] = "true" if _parse_bool(enable_thinking) else "false"
+        if use_max_completion_tokens is not None:
+            os.environ["QWEN_CHAT_USE_MAX_COMPLETION_TOKENS"] = (
+                "true" if _parse_bool(use_max_completion_tokens) else "false"
             )
         _update_config(
             OPTIMIZER_CONFIG,
             "optimizer",
             base_url=optimizer_base_url if optimizer_base_url is not None else base_url,
             api_key=optimizer_api_key if optimizer_api_key is not None else api_key,
-            temperature=(
-                optimizer_temperature
-                if optimizer_temperature is not None
-                else temperature
-            ),
-            timeout_seconds=(
-                optimizer_timeout_seconds
-                if optimizer_timeout_seconds is not None
-                else timeout_seconds
-            ),
+            temperature=(optimizer_temperature if optimizer_temperature is not None else temperature),
+            timeout_seconds=(optimizer_timeout_seconds if optimizer_timeout_seconds is not None else timeout_seconds),
             max_tokens=optimizer_max_tokens if optimizer_max_tokens is not None else max_tokens,
-            enable_thinking=(
-                optimizer_enable_thinking
-                if optimizer_enable_thinking is not None
-                else enable_thinking
+            enable_thinking=(optimizer_enable_thinking if optimizer_enable_thinking is not None else enable_thinking),
+            use_max_completion_tokens=(
+                optimizer_use_max_completion_tokens
+                if optimizer_use_max_completion_tokens is not None
+                else use_max_completion_tokens
             ),
         )
         _update_config(
@@ -287,16 +314,13 @@ def configure_qwen_chat(
             base_url=target_base_url if target_base_url is not None else base_url,
             api_key=target_api_key if target_api_key is not None else api_key,
             temperature=target_temperature if target_temperature is not None else temperature,
-            timeout_seconds=(
-                target_timeout_seconds
-                if target_timeout_seconds is not None
-                else timeout_seconds
-            ),
+            timeout_seconds=(target_timeout_seconds if target_timeout_seconds is not None else timeout_seconds),
             max_tokens=target_max_tokens if target_max_tokens is not None else max_tokens,
-            enable_thinking=(
-                target_enable_thinking
-                if target_enable_thinking is not None
-                else enable_thinking
+            enable_thinking=(target_enable_thinking if target_enable_thinking is not None else enable_thinking),
+            use_max_completion_tokens=(
+                target_use_max_completion_tokens
+                if target_use_max_completion_tokens is not None
+                else use_max_completion_tokens
             ),
         )
 
@@ -311,6 +335,7 @@ def _update_config(
     timeout_seconds: float | str | None = None,
     max_tokens: int | str | None = None,
     enable_thinking: bool | str | None = None,
+    use_max_completion_tokens: bool | str | None = None,
 ) -> None:
     env_prefix = role.upper()
     if base_url is not None:
@@ -321,7 +346,7 @@ def _update_config(
         os.environ[f"{env_prefix}_QWEN_CHAT_API_KEY"] = config.api_key
     if temperature is not None:
         raw = str(temperature).strip()
-        config.temperature = float(raw) if raw else None
+        config.temperature = None if raw.lower() in _OMIT_SENTINELS else float(raw)
         os.environ[f"{env_prefix}_QWEN_CHAT_TEMPERATURE"] = raw
     if timeout_seconds is not None:
         config.timeout_seconds = float(timeout_seconds)
@@ -331,8 +356,11 @@ def _update_config(
         os.environ[f"{env_prefix}_QWEN_CHAT_MAX_TOKENS"] = str(max_tokens)
     if enable_thinking is not None:
         config.enable_thinking = _parse_bool(enable_thinking)
-        os.environ[f"{env_prefix}_QWEN_CHAT_ENABLE_THINKING"] = (
-            "true" if config.enable_thinking else "false"
+        os.environ[f"{env_prefix}_QWEN_CHAT_ENABLE_THINKING"] = "true" if config.enable_thinking else "false"
+    if use_max_completion_tokens is not None:
+        config.use_max_completion_tokens = _parse_bool(use_max_completion_tokens)
+        os.environ[f"{env_prefix}_QWEN_CHAT_USE_MAX_COMPLETION_TOKENS"] = (
+            "true" if config.use_max_completion_tokens else "false"
         )
 
 

skillopt_sleep/__main__.py

@@ -13,7 +13,7 @@ Common flags:
     --max-tasks N       cap mined tasks per run
     --target-skill-path PATH explicit live SKILL.md to stage/adopt
     --tasks-file PATH   reviewed TaskRecord JSON file to replay instead of harvesting
-    --backend mock|claude|codex|copilot
+    --backend mock|claude|codex|copilot|handoff
     --source claude|codex|auto
     --model NAME
     --lookback-hours N
@@ -69,7 +69,8 @@ def _report_payload(rep, outcome) -> Dict[str, Any]:
 def _add_common(p: argparse.ArgumentParser) -> None:
     p.add_argument("--project", default="")
     p.add_argument("--scope", default="", choices=["", "all", "invoked"])
-    p.add_argument("--backend", default="", choices=["", "mock", "claude", "codex", "copilot"])
+    p.add_argument("--backend", default="",
+                   choices=["", "mock", "claude", "codex", "copilot", "handoff"])
     p.add_argument("--model", default="")
     p.add_argument("--codex-path", default="", help="path to the real @openai/codex binary")
     p.add_argument("--claude-home", default="", help="override ~/.claude (also isolates state)")
@@ -156,7 +157,14 @@ def cmd_run(args, dry: bool = False) -> int:
                 file=sys.stderr,
             )
             return 2
+    if cfg.get("backend", "mock") == "handoff":
+        return _run_handoff(cfg, args, seed_tasks=tasks, task_meta=task_meta, dry=dry)
     outcome = run_sleep_cycle(cfg, seed_tasks=tasks, dry_run=dry)
+    _print_run_report(outcome, args, task_meta)
+    return 0
+
+
+def _print_run_report(outcome, args, task_meta: Dict[str, Any]) -> None:
     rep = outcome.report
     if args.json:
         payload = _report_payload(rep, outcome)
@@ -180,6 +188,202 @@ def cmd_run(args, dry: bool = False) -> int:
                 print("[sleep] review it, then: python -m skillopt_sleep adopt")
         if outcome.adopted:
             print(f"[sleep] auto-adopted: {', '.join(outcome.adopted_paths)}")
+
+
+def _handoff_dir_for(cfg) -> str:
+    project = cfg.get("invoked_project") or os.getcwd()
+    return os.environ.get("SKILLOPT_SLEEP_HANDOFF_DIR", "") or os.path.join(
+        project, ".skillopt-sleep-handoff"
+    )
+
+
+def _redact_deep(obj):
+    """Redact secret-looking substrings in every string of a JSON-like tree."""
+    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
+
+
+def _flush_handoff(backend, args) -> int:
+    prompts_path = backend.flush_pending()
+    if args.json:
+        print(json.dumps({
+            "handoff_pending": len(backend.pending),
+            "prompts": prompts_path,
+            "answers_dir": backend.answers_dir,
+        }, ensure_ascii=False, indent=2))
+    else:
+        print(f"[sleep] handoff: {len(backend.pending)} model call(s) need answers")
+        print(f"[sleep] prompts: {prompts_path}")
+        print(f"[sleep] write each raw answer to {backend.answers_dir}/<id>.md, "
+              "then re-run this exact command to resume")
+    return 3
+
+
+def _handoff_mine_and_pin(cfg, args, backend, snapshot: str, dry: bool):
+    """Harvest + mine with the same knobs as run_sleep_cycle (harvest window,
+    target-skill filter, candidate-limit bump, LLM mining — routed through the
+    handoff files like every other model call), then pin the result to
+    ``tasks.json``. Session digests are pinned too, so the sessions created
+    while answering prompts cannot change what gets mined between rounds.
+
+    Returns ``(exit_code, tasks)``; ``tasks is None`` means exit now.
+    """
+    import time
+
+    from skillopt_sleep.handoff_backend import PendingCalls
+    from skillopt_sleep.state import SleepState, _now_iso
+    from skillopt_sleep.types import SessionDigest
+
+    project = cfg.get("invoked_project") or os.getcwd()
+    state = SleepState.load(cfg.state_path)
+    started = _now_iso()
+
+    digests_path = os.path.join(backend.handoff_dir, "digests.json")
+    digests = None
+    if os.path.exists(digests_path):
+        try:
+            with open(digests_path, encoding="utf-8") as f:
+                raw = json.load(f)
+            known = set(SessionDigest.__dataclass_fields__)
+            digests = [SessionDigest(**{k: v for k, v in d.items() if k in known})
+                       for d in raw]
+        except Exception:
+            # Corrupted/truncated pin (e.g. an interrupted earlier round):
+            # fall back to a fresh harvest instead of crashing the run.
+            print("[sleep] handoff: digests.json unreadable — re-harvesting",
+                  file=sys.stderr)
+            digests = None
+    if digests is None:
+        since = state.last_harvest_for(project)
+        lookback_hours = cfg.get("lookback_hours", 72)
+        if since is None and lookback_hours and lookback_hours > 0:
+            since = _now_iso(time.time() - lookback_hours * 3600)
+        max_tasks = cfg.get("max_tasks_per_night", 40)
+        session_limit = cfg.get("max_sessions_per_night", 0) or max_tasks * 3
+        digests = harvest_for_config(cfg, since_iso=since, limit=session_limit)
+        os.makedirs(backend.handoff_dir, exist_ok=True)
+        with open(digests_path, "w", encoding="utf-8") as f:
+            json.dump(_redact_deep([d.to_dict() for d in digests]), f,
+                      ensure_ascii=False, indent=2)
+
+    max_tasks = cfg.get("max_tasks_per_night", 40)
+    session_limit = cfg.get("max_sessions_per_night", 0) or max_tasks * 3
+    target_skill_path = cfg.managed_skill_path() if cfg.get("target_skill_path", "") else ""
+    target_skill_text = _read_text(target_skill_path) if target_skill_path else ""
+    candidate_limit = max_tasks
+    if cfg.get("target_task_filter", True) and target_skill_text:
+        candidate_limit = max(max_tasks, max_tasks * 3)
+    llm_miner = None
+    if cfg.get("llm_mine", True):
+        try:
+            from skillopt_sleep.llm_miner import make_llm_miner
+            llm_miner = make_llm_miner(
+                backend, max_sessions=session_limit, max_tasks=candidate_limit,
+            )
+        except Exception:
+            llm_miner = None
+    try:
+        tasks = mine(
+            digests,
+            max_tasks=max_tasks,
+            candidate_limit=candidate_limit,
+            holdout_fraction=cfg.get("holdout_fraction", 0.34),
+            seed=cfg.get("seed", 42),
+            llm_miner=llm_miner,
+            target_skill_text=target_skill_text,
+            target_skill_path=target_skill_path,
+        )
+    except PendingCalls:
+        tasks = []
+    if backend.pending:
+        # LLM mining needs answers before the task set can be pinned.
+        return _flush_handoff(backend, args), None
+    if not tasks:
+        print("[sleep] handoff: no tasks mined — nothing to consolidate")
+        if not dry:
+            # Advance the harvest window like run_sleep_cycle's no-tasks
+            # branch, or every later run re-scans the same stale window.
+            state.set_last_harvest(project, started)
+            state.save()
+            # Drop the pinned (often empty) digests so the NEXT run re-harvests
+            # instead of reloading a []-pin and never calling harvest again.
+            try:
+                if os.path.exists(digests_path):
+                    os.remove(digests_path)
+            except OSError:
+                pass
+        return 0, None
+    payload = make_tasks_payload(
+        tasks,
+        project=project,
+        transcript_source=cfg.get("transcript_source", ""),
+        n_sessions=len(digests),
+        target_skill_path=target_skill_path,
+    )
+    # 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))
+    print(f"[sleep] handoff: pinned {len(tasks)} tasks -> {snapshot}")
+    return 0, tasks
+
+
+def _run_handoff(cfg, args, *, seed_tasks, task_meta: Dict[str, Any], dry: bool) -> int:
+    """Drive the handoff backend: run until model calls are needed, then
+    write the prompt batch and exit 3; on a fully-answered run, finish
+    normally. Session digests and mined tasks are pinned under the handoff
+    dir on the first rounds so wall-clock time between rounds (including
+    the very sessions that answer the prompts) cannot change the task set
+    and invalidate earlier answers.
+    """
+    from skillopt_sleep.handoff_backend import HandoffBackend, PendingCalls
+
+    hdir = _handoff_dir_for(cfg)
+    backend = HandoffBackend(model=cfg.get("model", ""), handoff_dir=hdir)
+    tasks = seed_tasks
+    if tasks is None:
+        snapshot = os.path.join(hdir, "tasks.json")
+        tasks = None
+        if os.path.exists(snapshot):
+            try:
+                tasks, _meta = load_tasks_file(
+                    snapshot,
+                    holdout_fraction=cfg.get("holdout_fraction", 0.34),
+                    seed=cfg.get("seed", 42),
+                )
+            except (ValueError, json.JSONDecodeError, OSError):
+                # Interrupted/corrupt pin from an earlier round: re-mine instead
+                # of crashing the run.
+                print("[sleep] handoff: tasks.json unreadable — re-mining",
+                      file=sys.stderr)
+                tasks = None
+        if tasks is None:
+            rc, tasks = _handoff_mine_and_pin(cfg, args, backend, snapshot, dry)
+            if tasks is None:
+                return rc
+    outcome = None
+    try:
+        outcome = run_sleep_cycle(cfg, seed_tasks=tasks, dry_run=dry, backend=backend)
+    except PendingCalls:
+        pass
+    if backend.pending:
+        return _flush_handoff(backend, args)
+    _print_run_report(outcome, args, task_meta)
+    # A completed real run ends the night: archive the handoff dir so the
+    # next night re-harvests instead of replaying the pinned snapshot.
+    if not dry and outcome.staging_dir and os.path.isdir(hdir):
+        import time
+        done = f"{hdir}.night{outcome.report.night}.done"
+        if os.path.exists(done):
+            done = f"{done}.{int(time.time())}"
+        os.rename(hdir, done)
+        print(f"[sleep] handoff: archived round data -> {done}")
     return 0
 
 

skillopt_sleep/backend.py

@@ -29,6 +29,12 @@ from typing import Any, Dict, List, Optional, Tuple
 
 from skillopt_sleep.types import EditRecord, ReplayResult, TaskRecord
 
+# On Windows, console-attached children (cmd.exe shims, python) allocate a
+# visible console window when the parent has none — a nightly cycle making
+# hundreds of CLI calls strobes cmd windows and steals focus from the user.
+# CREATE_NO_WINDOW suppresses that; harmless 0 elsewhere.
+_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0
+
 
 def skill_hash(content: str) -> str:
     import hashlib
@@ -555,7 +561,12 @@ class ClaudeCliBackend(CliBackend):
     def __init__(self, model: str = "", claude_path: str = "claude", timeout: int = 180) -> None:
         super().__init__(model=model or os.environ.get("SKILLOPT_SLEEP_CLAUDE_MODEL", "") or "sonnet",
                          timeout=timeout)
-        self.claude_path = claude_path
+        # On Windows the npm-installed `claude` is a .cmd shim; CreateProcess
+        # cannot resolve it by bare name (WinError 2), so every call would
+        # silently return "" and the whole cycle scores 0.0. shutil.which
+        # honors PATHEXT and returns the full claude.CMD path.
+        import shutil as _shutil
+        self.claude_path = _shutil.which(claude_path) or claude_path
 
     # Known CLI error prefixes that indicate auth or config failures.
     # When detected, we log a warning so the user doesn't mistake a
@@ -614,13 +625,21 @@ class ClaudeCliBackend(CliBackend):
         ]
         if self.model:
             cmd += ["--model", self.model]
-        cmd += ["--", prompt]
+        # Prompt goes via stdin, not argv: the Windows .cmd shim routes through
+        # cmd.exe whose command line caps at ~8K chars — reflect/judge prompts
+        # exceed that. `claude -p` with no positional prompt reads stdin.
         clean_cwd = tempfile.mkdtemp(prefix="skillopt_sleep_claude_")
         try:
             proc = subprocess.run(
-                cmd, capture_output=True, text=True, timeout=self.timeout, cwd=clean_cwd,
+                cmd, capture_output=True, creationflags=_NO_WINDOW, text=True, timeout=self.timeout, cwd=clean_cwd,
+                input=prompt,
+            )
+        except Exception as exc:
+            import logging
+            self.last_call_error = f"Claude CLI spawn failed: {exc}"
+            logging.getLogger("skillopt_sleep").warning(
+                "Claude CLI could not be executed: %s", exc,
             )
-        except Exception:
             return ""
         finally:
             try:
@@ -677,14 +696,19 @@ class ClaudeCliBackend(CliBackend):
             ]
             if self.model:
                 cmd += ["--model", self.model]
-            cmd += ["--", prompt]
             try:
                 proc = subprocess.run(
-                    cmd, capture_output=True, text=True, timeout=self.timeout, cwd=work,
+                    cmd, capture_output=True, creationflags=_NO_WINDOW, text=True, timeout=self.timeout, cwd=work,
+                    input=prompt,
                 )
                 resp = (proc.stdout or "").strip()
                 self._detect_cli_error(resp, proc.stderr or "")
-            except Exception:
+            except Exception as exc:
+                import logging
+                self.last_call_error = f"Claude CLI spawn failed: {exc}"
+                logging.getLogger("skillopt_sleep").warning(
+                    "Claude CLI could not be executed: %s", exc,
+                )
                 resp = ""
             self._tokens += len(prompt) // 4 + len(resp) // 4
             called: List[str] = []
@@ -779,6 +803,7 @@ class CodexCliBackend(CliBackend):
                 proc = subprocess.run(
                     cmd,
                     capture_output=True,
+                    creationflags=_NO_WINDOW,
                     text=True,
                     timeout=self.timeout,
                     cwd=self.project_dir or None,
@@ -896,7 +921,7 @@ class CodexCliBackend(CliBackend):
             self.last_call_error = ""
             proc = None
             try:
-                proc = subprocess.run(cmd, capture_output=True, text=True, timeout=self.timeout, cwd=work)
+                proc = subprocess.run(cmd, capture_output=True, creationflags=_NO_WINDOW, text=True, timeout=self.timeout, cwd=work)
             except subprocess.TimeoutExpired:
                 self.last_call_error = f"codex exec (tools) timed out after {self.timeout}s"
             except Exception as exc:  # noqa: BLE001
@@ -1006,7 +1031,7 @@ class CopilotCliBackend(CliBackend):
             env["COPILOT_HOME"] = self.copilot_home
         try:
             proc = subprocess.run(
-                cmd, capture_output=True, text=True, timeout=self.timeout, cwd=clean_cwd,
+                cmd, capture_output=True, creationflags=_NO_WINDOW, text=True, timeout=self.timeout, cwd=clean_cwd,
                 encoding="utf-8", errors="replace", env=env,
             )
         except Exception:
@@ -1119,7 +1144,7 @@ class CopilotCliBackend(CliBackend):
             resp = ""
             try:
                 proc = subprocess.run(
-                    cmd, capture_output=True, text=True, encoding="utf-8",
+                    cmd, capture_output=True, creationflags=_NO_WINDOW, text=True, encoding="utf-8",
                     errors="replace", timeout=self.timeout, cwd=work, env=env,
                 )
                 resp = self._parse_jsonl_response(proc.stdout or "")
@@ -1390,6 +1415,13 @@ def get_backend(
         return AzureResponsesBackend(deployment=model, endpoints=eps)
     if n in {"copilot", "github_copilot", "copilot_cli", "gh_copilot"}:
         return CopilotCliBackend(model=model)
+    if n in {"handoff", "session", "file"}:
+        # Lazy import: handoff_backend imports CliBackend from this module.
+        from skillopt_sleep.handoff_backend import HandoffBackend
+        hdir = os.environ.get("SKILLOPT_SLEEP_HANDOFF_DIR", "") or os.path.join(
+            project_dir or os.getcwd(), ".skillopt-sleep-handoff"
+        )
+        return HandoffBackend(model=model, handoff_dir=hdir)
     return MockBackend()
 
 

skillopt_sleep/cycle.py

@@ -14,7 +14,7 @@ import sys
 from dataclasses import dataclass
 from typing import List, Optional
 
-from skillopt_sleep.backend import get_backend
+from skillopt_sleep.backend import Backend, get_backend
 from skillopt_sleep.config import SleepConfig, load_config
 from skillopt_sleep.dream import dream_consolidate
 from skillopt_sleep.harvest_sources import harvest_for_config
@@ -94,6 +94,7 @@ def run_sleep_cycle(
     seed_tasks: Optional[List[TaskRecord]] = None,
     dry_run: bool = False,
     clock: Optional[float] = None,
+    backend: Optional[Backend] = None,
 ) -> CycleOutcome:
     """Run one full sleep cycle and return the outcome.
 
@@ -104,6 +105,8 @@ def run_sleep_cycle(
         inject a known persona instead of harvesting ~/.claude).
     dry_run : harvest+mine+replay but DO NOT stage/adopt (report only).
     clock : fixed epoch seconds for deterministic timestamps in tests.
+    backend : optional pre-built Backend; the handoff driver passes one so
+        it can inspect the backend's pending calls after the run.
     """
     cfg = cfg or load_config()
     state = SleepState.load(cfg.state_path)
@@ -111,7 +114,7 @@ def run_sleep_cycle(
     project = _project_paths(cfg)
     started = _now_iso(clock)
 
-    backend = get_backend(
+    backend = backend or get_backend(
         cfg.get("backend", "mock"),
         model=cfg.get("model", ""),
         codex_path=cfg.get("codex_path", ""),

skillopt_sleep/handoff_backend.py

@@ -0,0 +1,173 @@
+"""SkillOpt-Sleep — handoff backend (session-executed model calls).
+
+Runs the sleep cycle WITHOUT spawning any model subprocess or API call.
+Every intelligent operation (attempt / judge / reflect) is turned into a
+prompt file that an interactive agent session answers between engine runs:
+
+    run 1: the engine executes the deterministic stages; every model call
+           it needs is recorded as a pending prompt; the run stops and
+           writes PROMPTS.md + pending.json into the handoff directory.
+    you:   answer each prompt (each in a FRESH context, so the session's
+           own history cannot contaminate the held-out gate) and write the
+           raw answer text to answers/<id>.md.
+    run 2: the engine re-runs; answered prompts resolve from answers/, the
+           cycle advances to the next model-dependent stage, and either
+           finishes or writes the next PROMPTS.md batch.
+
+Resume needs no serialized engine state: harvest -> mine -> replay is
+deterministic, so re-running regenerates identical prompts and the answers
+directory acts as a persistent, cross-run call cache. A prompt that embeds
+a still-unanswered response (detected via the pending sentinel) aborts the
+run immediately so placeholder text never propagates into scores, edits,
+or staging. A typical night converges in 3-6 rounds: baseline attempts ->
+reflect -> candidate re-scoring per accepted edit.
+
+Limitations (v1): `dream_rollouts > 1` yields no contrastive spread (the
+same prompt maps to the same answer file), and tool-loop tasks fall back
+to the base single-shot 'TOOL_CALL: <name>' marker convention.
+"""
+from __future__ import annotations
+
+import json
+import os
+import threading
+from typing import Dict
+
+from skillopt_sleep.backend import CliBackend, skill_hash
+
+PENDING_SENTINEL_PREFIX = "[[SKILLOPT-SLEEP-PENDING:"
+PENDING_SENTINEL_SUFFIX = "]]"
+
+# reflect() appends this when a reply fails to parse; with a placeholder
+# reply the retry is a dependent call, not a genuinely new question.
+_REFLECT_RETRY_MARKER = "your previous reply was not valid JSON"
+
+PROMPTS_FILENAME = "PROMPTS.md"
+PENDING_FILENAME = "pending.json"
+
+
+class PendingCalls(RuntimeError):
+    """The cycle cannot advance until pending prompts are answered."""
+
+    def __init__(self, pending: Dict[str, Dict[str, object]]):
+        self.pending = dict(pending)
+        super().__init__(
+            f"{len(self.pending)} model call(s) awaiting handoff answers"
+        )
+
+
+class HandoffBackend(CliBackend):
+    """Backend that outsources every model call to prompt/answer files.
+
+    ``_call`` resolves a prompt from ``answers/<sha256[:16]>.md`` when the
+    answer exists; otherwise it records the prompt as pending and returns a
+    sentinel placeholder so independent calls in the same phase can still
+    be collected into one batch. Any call whose prompt was BUILT FROM a
+    placeholder raises :class:`PendingCalls` — that call depends on answers
+    the user has not provided yet, so continuing would only mint garbage.
+    """
+
+    name = "handoff"
+
+    def __init__(self, model: str = "", handoff_dir: str = "") -> None:
+        super().__init__(model=model, timeout=0)
+        self.handoff_dir = os.path.abspath(
+            handoff_dir or os.path.join(os.getcwd(), ".skillopt-sleep-handoff")
+        )
+        self.answers_dir = os.path.join(self.handoff_dir, "answers")
+        os.makedirs(self.answers_dir, exist_ok=True)
+        # key -> {"prompt": str, "max_tokens": int}, insertion-ordered
+        self.pending: Dict[str, Dict[str, object]] = {}
+        self._lock = threading.Lock()
+
+    # ── prompt/answer plumbing ────────────────────────────────────────────
+    def answer_path(self, key: str) -> str:
+        return os.path.join(self.answers_dir, f"{key}.md")
+
+    def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
+        if PENDING_SENTINEL_PREFIX in prompt:
+            # Built from a still-pending response — dependent call.
+            raise PendingCalls(self.pending)
+        if _REFLECT_RETRY_MARKER in prompt and self.pending:
+            # Retry of a reflect whose first reply is the placeholder.
+            raise PendingCalls(self.pending)
+        key = skill_hash(prompt)
+        path = self.answer_path(key)
+        if os.path.exists(path):
+            with open(path, encoding="utf-8") as f:
+                return f.read().strip()
+        with self._lock:
+            self.pending[key] = {"prompt": prompt, "max_tokens": max_tokens}
+        return f"{PENDING_SENTINEL_PREFIX}{key}{PENDING_SENTINEL_SUFFIX}"
+
+    # ── handoff file emission ─────────────────────────────────────────────
+    def flush_pending(self) -> str:
+        """Write PROMPTS.md (human/agent-readable) + pending.json (machine).
+
+        Prompts can themselves contain markdown fences, so PROMPTS.md
+        delimits each prompt with BEGIN/END marker lines instead of fences.
+        Returns the PROMPTS.md path.
+        """
+        from skillopt_sleep.staging import redact_secrets
+
+        os.makedirs(self.handoff_dir, exist_ok=True)
+        with self._lock:
+            items = list(self.pending.items())
+        payload = {
+            "format": "skillopt_sleep.handoff.v1",
+            "answers_dir": self.answers_dir,
+            "pending": [
+                {
+                    "id": key,
+                    "answer_file": self.answer_path(key),
+                    "max_tokens": item["max_tokens"],
+                    "prompt": redact_secrets(str(item["prompt"])),
+                }
+                for key, item in items
+            ],
+        }
+        with open(os.path.join(self.handoff_dir, PENDING_FILENAME), "w",
+                  encoding="utf-8") as f:
+            json.dump(payload, f, ensure_ascii=False, indent=2)
+            f.write("\n")
+
+        lines = [
+            "# SkillOpt-Sleep — pending model calls (handoff)",
+            "",
+            f"{len(items)} prompt(s) below need answers before the sleep "
+            "cycle can continue.",
+            "",
+            "For EACH prompt:",
+            "",
+            "1. Answer it in a FRESH context (e.g. a subagent with no",
+            "   conversation history). Do NOT let the current session's",
+            "   context, the other prompts in this file, or the optimization",
+            "   run itself leak into the answer — that contaminates the",
+            "   held-out validation gate.",
+            "2. Write ONLY the raw answer text (no commentary, no code",
+            "   fences) to the prompt's answer file.",
+            "",
+            "When every answer file exists, re-run the same engine command",
+            "(`python -m skillopt_sleep run --backend handoff ...`); it",
+            "resumes automatically from the answers directory.",
+            "",
+        ]
+        for i, (key, item) in enumerate(items, start=1):
+            lines += [
+                "---",
+                "",
+                f"## Prompt {i} of {len(items)}",
+                "",
+                f"- id: `{key}`",
+                f"- answer file: `answers/{key}.md`",
+                f"- suggested max tokens: {item['max_tokens']}",
+                "",
+                f"----- BEGIN PROMPT {key} -----",
+                redact_secrets(str(item["prompt"])),
+                f"----- END PROMPT {key} -----",
+                "",
+            ]
+        prompts_path = os.path.join(self.handoff_dir, PROMPTS_FILENAME)
+        with open(prompts_path, "w", encoding="utf-8") as f:
+            f.write("\n".join(lines))
+        return prompts_path

skillopt_sleep/harvest.py

@@ -109,6 +109,13 @@ def _is_meta_prompt(text: str) -> bool:
         return True
     if t.startswith("[Pasted text") or t.startswith("Caveat:"):
         return True
+    # Expanded slash-command prompts (Claude Code injects the full command
+    # body as a user message, wrapped in <command-message>/<command-name>
+    # tags or rendered as "# /<name> — ..." headers). Not a user intent.
+    if "<command-message>" in t[:200] or "<command-name>" in t[:200]:
+        return True
+    if t.startswith("# /"):
+        return True
     return False
 
 
@@ -131,6 +138,31 @@ _REPLAY_PROMPT_MARKERS = (
 )
 
 
+# Sessions written by OTHER tools' sub-agents (memory observers, critic
+# sub-agents, plugin self-invocations). These are multi-turn, so the
+# single-turn heuristic in _is_headless_replay never catches them. If the
+# FIRST user prompt matches any marker, the whole session is
+# machine-generated, not a real user task.
+_AGENT_SESSION_MARKERS = (
+    "You are a Claude-Mem",              # claude-mem observer agent
+    "Hello memory agent",                # claude-mem observer continuation
+    "You are driving **SkillOpt-Sleep**",  # this plugin's own command body
+) + tuple(
+    # users can add markers for their own tools' agent prompts
+    p.strip()
+    for p in os.environ.get("SKILLOPT_SLEEP_AGENT_MARKERS", "").split(",")
+    if p.strip()
+)
+
+
+def _is_agent_session(digest: "SessionDigest") -> bool:
+    """Detect transcripts written by other tools' sub-agents (see markers)."""
+    if not digest.user_prompts:
+        return False
+    first = digest.user_prompts[0]
+    return any(marker in first for marker in _AGENT_SESSION_MARKERS)
+
+
 def _is_headless_replay(digest: "SessionDigest") -> bool:
     """Detect sessions created by the engine's own headless replay calls.
 
@@ -279,8 +311,12 @@ def harvest(
 
     paths: List[str] = []
     for root, _dirs, files in os.walk(transcripts_dir):
+        # Sub-agent sidechain transcripts (<session>/subagents/agent-*.jsonl)
+        # are Claude-authored prompts, not user tasks — never harvest them.
+        if os.path.basename(root) == "subagents":
+            continue
         for fn in files:
-            if fn.endswith(".jsonl"):
+            if fn.endswith(".jsonl") and not fn.startswith("agent-"):
                 paths.append(os.path.join(root, fn))
     # newest first by mtime
     paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
@@ -291,6 +327,8 @@ def harvest(
             continue
         if _is_headless_replay(d):
             continue  # Issue #62: skip engine's own headless replay sessions
+        if _is_agent_session(d):
+            continue  # skip other tools' sub-agent transcripts (claude-mem etc.)
         if not _project_matches(d.project or "", scope, invoked_project):
             continue
         if since_iso and d.ended_at and d.ended_at < since_iso:

tests/test_gate.py

@@ -0,0 +1,286 @@
+"""Tests for skillopt.evaluation.gate — the validation gate decision function.
+
+The gate is the optimizer's model-selection / early-stopping core: given a
+candidate skill's score, it decides whether to accept it as the new current
+skill and whether it becomes the new best-so-far. These are pure functions,
+so they can be exercised directly without any LLM or rollout.
+"""
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from skillopt.evaluation.gate import (
+    GateResult,
+    evaluate_gate,
+    select_gate_score,
+)
+
+
+class TestSelectGateScore:
+    """select_gate_score — project (hard, soft) onto a single comparison metric."""
+
+    def test_hard_metric_returns_hard(self) -> None:
+        assert select_gate_score(0.8, 0.3, "hard") == 0.8
+
+    def test_soft_metric_returns_soft(self) -> None:
+        assert select_gate_score(0.8, 0.3, "soft") == 0.3
+
+    def test_default_metric_is_hard(self) -> None:
+        assert select_gate_score(0.42, 0.99) == 0.42
+
+    def test_mixed_metric_default_weight(self) -> None:
+        # (1 - 0.5) * 1.0 + 0.5 * 0.0 == 0.5
+        assert select_gate_score(1.0, 0.0, "mixed") == pytest.approx(0.5)
+
+    def test_mixed_metric_custom_weight(self) -> None:
+        # (1 - 0.25) * 0.8 + 0.25 * 0.4 == 0.7
+        assert select_gate_score(0.8, 0.4, "mixed", 0.25) == pytest.approx(0.7)
+
+    def test_mixed_weight_zero_equals_hard(self) -> None:
+        assert select_gate_score(0.8, 0.3, "mixed", 0.0) == pytest.approx(0.8)
+
+    def test_mixed_weight_one_equals_soft(self) -> None:
+        assert select_gate_score(0.8, 0.3, "mixed", 1.0) == pytest.approx(0.3)
+
+    def test_mixed_weight_clamped_above_one(self) -> None:
+        """Out-of-range weight is clamped to 1.0 (→ pure soft)."""
+        assert select_gate_score(0.8, 0.3, "mixed", 5.0) == pytest.approx(0.3)
+
+    def test_mixed_weight_clamped_below_zero(self) -> None:
+        """Negative weight is clamped to 0.0 (→ pure hard)."""
+        assert select_gate_score(0.8, 0.3, "mixed", -2.0) == pytest.approx(0.8)
+
+    def test_returns_float(self) -> None:
+        assert isinstance(select_gate_score(1, 0, "hard"), float)
+
+    def test_unknown_metric_raises(self) -> None:
+        with pytest.raises(ValueError, match="unknown gate metric"):
+            select_gate_score(0.5, 0.5, "rouge")  # type: ignore[arg-type]
+
+
+class TestEvaluateGateAcceptNewBest:
+    """evaluate_gate — candidate beats both current and best."""
+
+    def test_accept_new_best_action_and_state(self) -> None:
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.9,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.5,
+            best_step=3,
+            global_step=7,
+        )
+        assert result.action == "accept_new_best"
+        assert result.current_skill == "CAND"
+        assert result.current_score == pytest.approx(0.9)
+        assert result.best_skill == "CAND"
+        assert result.best_score == pytest.approx(0.9)
+        assert result.best_step == 7  # updated to the accepting step
+
+
+class TestEvaluateGateAccept:
+    """evaluate_gate — candidate beats current but not best.
+
+    This branch is only reachable when ``current_score < best_score``; it
+    advances the current skill without disturbing the best-so-far checkpoint.
+    """
+
+    def test_accept_updates_current_only(self) -> None:
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.6,
+            current_skill="CURR",
+            current_score=0.4,
+            best_skill="BEST",
+            best_score=0.8,
+            best_step=2,
+            global_step=9,
+        )
+        assert result.action == "accept"
+        assert result.current_skill == "CAND"
+        assert result.current_score == pytest.approx(0.6)
+        # best-so-far is preserved, including its step
+        assert result.best_skill == "BEST"
+        assert result.best_score == pytest.approx(0.8)
+        assert result.best_step == 2
+
+    def test_tie_with_best_but_above_current_accepts(self) -> None:
+        """cand == best (not strictly greater) but > current → accept, not new best."""
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.8,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.8,
+            best_step=1,
+            global_step=4,
+        )
+        assert result.action == "accept"
+        assert result.current_skill == "CAND"
+        assert result.best_skill == "BEST"
+        assert result.best_score == pytest.approx(0.8)
+        assert result.best_step == 1
+
+
+class TestEvaluateGateReject:
+    """evaluate_gate — candidate does not beat current."""
+
+    def test_reject_below_current(self) -> None:
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.3,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.8,
+            best_step=2,
+            global_step=6,
+        )
+        assert result.action == "reject"
+        assert result.current_skill == "CURR"
+        assert result.current_score == pytest.approx(0.5)
+        assert result.best_skill == "BEST"
+        assert result.best_score == pytest.approx(0.8)
+        assert result.best_step == 2
+
+    def test_tie_with_current_rejects(self) -> None:
+        """Strict inequality: cand == current is rejected (no lateral moves)."""
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.5,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.5,
+            best_step=0,
+            global_step=3,
+        )
+        assert result.action == "reject"
+        assert result.current_skill == "CURR"
+        assert result.best_skill == "BEST"
+
+
+class TestEvaluateGateMetrics:
+    """evaluate_gate — non-hard metrics drive the comparison via cand_soft."""
+
+    def test_soft_metric_uses_cand_soft(self) -> None:
+        # High hard, low soft: under 'soft' the candidate must be rejected.
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.95,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.5,
+            best_step=0,
+            global_step=1,
+            cand_soft=0.2,
+            metric="soft",
+        )
+        assert result.action == "reject"
+
+    def test_mixed_metric_uses_weighted_score(self) -> None:
+        # mixed w=0.5: (0.5 * 1.0) + (0.5 * 0.6) == 0.8 > current 0.5 and best 0.5
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=1.0,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.5,
+            best_step=0,
+            global_step=2,
+            cand_soft=0.6,
+            metric="mixed",
+            mixed_weight=0.5,
+        )
+        assert result.action == "accept_new_best"
+        assert result.current_score == pytest.approx(0.8)
+        assert result.best_score == pytest.approx(0.8)
+
+    def test_default_metric_ignores_soft(self) -> None:
+        """Default metric is 'hard'; cand_soft must not affect the decision."""
+        result = evaluate_gate(
+            candidate_skill="CAND",
+            cand_hard=0.9,
+            current_skill="CURR",
+            current_score=0.5,
+            best_skill="BEST",
+            best_score=0.5,
+            best_step=0,
+            global_step=1,
+            cand_soft=0.0,
+        )
+        assert result.action == "accept_new_best"
+        assert result.current_score == pytest.approx(0.9)
+
+
+class TestGateResult:
+    """GateResult — immutable outcome dataclass."""
+
+    def test_fields(self) -> None:
+        result = GateResult(
+            action="accept",
+            current_skill="c",
+            current_score=0.5,
+            best_skill="b",
+            best_score=0.9,
+            best_step=4,
+        )
+        assert result.action == "accept"
+        assert result.current_skill == "c"
+        assert result.current_score == 0.5
+        assert result.best_skill == "b"
+        assert result.best_score == 0.9
+        assert result.best_step == 4
+
+    def test_is_frozen(self) -> None:
+        result = GateResult(
+            action="reject",
+            current_skill="c",
+            current_score=0.0,
+            best_skill="b",
+            best_score=0.0,
+            best_step=0,
+        )
+        with pytest.raises(dataclasses.FrozenInstanceError):
+            result.current_score = 1.0  # type: ignore[misc]
+
+
+class TestGateInvariants:
+    """Behavioral invariants of the gate over a sequence of steps."""
+
+    def test_current_tracks_best_from_equal_start(self) -> None:
+        """When current == best at the start, every acceptance is a new best, so
+        the two stay locked together and the 'accept' branch is never taken.
+
+        This documents the trainer's ``s_cur``/``s_best`` usage: they are
+        initialized equal and updated only through this gate.
+        """
+        current_skill, current_score = "S0", 0.2
+        best_skill, best_score, best_step = "S0", 0.2, 0
+        for step, cand in enumerate([0.1, 0.5, 0.4, 0.7], start=1):
+            result = evaluate_gate(
+                candidate_skill=f"S{step}",
+                cand_hard=cand,
+                current_skill=current_skill,
+                current_score=current_score,
+                best_skill=best_skill,
+                best_score=best_score,
+                best_step=best_step,
+                global_step=step,
+            )
+            current_skill, current_score = result.current_skill, result.current_score
+            best_skill = result.best_skill
+            best_score = result.best_score
+            best_step = result.best_step
+            assert result.action in {"accept_new_best", "reject"}
+            assert current_score == best_score
+            assert current_skill == best_skill
+        assert best_score == pytest.approx(0.7)
+        assert best_step == 4

tests/test_handoff_backend.py

@@ -0,0 +1,259 @@
+"""Tests for the handoff backend: session-executed model calls via
+prompt/answer files, resumed across engine runs."""
+from __future__ import annotations
+
+import json
+import os
+import re
+import tempfile
+import unittest
+
+from skillopt_sleep.backend import get_backend
+from skillopt_sleep.config import load_config
+from skillopt_sleep.cycle import run_sleep_cycle
+from skillopt_sleep.handoff_backend import (
+    PENDING_SENTINEL_PREFIX,
+    HandoffBackend,
+    PendingCalls,
+)
+from skillopt_sleep.mine import assign_splits
+from skillopt_sleep.types import TaskRecord
+
+# The rule the simulated executor "learns"; once present in the skill the
+# executor answers arithmetic tasks correctly, mirroring MockBackend's
+# rule-gated model of reality.
+RULE = "Always answer with just the number."
+
+
+def _tasks():
+    ts = [
+        TaskRecord(
+            id=f"t{i}", project="/p",
+            intent=f"What is {i} + {i}?",
+            reference_kind="exact", reference=str(i + i),
+        )
+        for i in range(1, 7)
+    ]
+    return assign_splits(ts, holdout_fraction=0.34, seed=42)
+
+
+def _answer_pending(backend: HandoffBackend) -> None:
+    """Deterministic stand-in for the interactive session answering PROMPTS.md."""
+    for key, item in list(backend.pending.items()):
+        prompt = str(item["prompt"])
+        if "You are SkillOpt's optimizer" in prompt:
+            ans = json.dumps([{
+                "op": "add", "content": RULE,
+                "rationale": "outputs failed exact match; answer bare numbers",
+            }])
+        else:
+            m = re.search(r"What is (\d+) \+ (\d+)\?", prompt)
+            if m and RULE in prompt:
+                ans = str(int(m.group(1)) + int(m.group(2)))
+            else:
+                ans = "cannot say"
+        with open(backend.answer_path(key), "w", encoding="utf-8") as f:
+            f.write(ans)
+
+
+class TestHandoffBackendUnit(unittest.TestCase):
+    def test_miss_records_pending_and_returns_sentinel(self):
+        with tempfile.TemporaryDirectory() as hdir:
+            be = HandoffBackend(handoff_dir=hdir)
+            out = be._call("some prompt")
+            self.assertTrue(out.startswith(PENDING_SENTINEL_PREFIX))
+            self.assertEqual(len(be.pending), 1)
+
+    def test_answer_file_resolves_call(self):
+        with tempfile.TemporaryDirectory() as hdir:
+            be = HandoffBackend(handoff_dir=hdir)
+            key = be._call("what is 2+2?").split(":")[-1].rstrip("]")
+            with open(be.answer_path(key), "w", encoding="utf-8") as f:
+                f.write("4\n")
+            fresh = HandoffBackend(handoff_dir=hdir)
+            self.assertEqual(fresh._call("what is 2+2?"), "4")
+            self.assertEqual(len(fresh.pending), 0)
+
+    def test_dependent_prompt_raises(self):
+        with tempfile.TemporaryDirectory() as hdir:
+            be = HandoffBackend(handoff_dir=hdir)
+            placeholder = be._call("first question")
+            with self.assertRaises(PendingCalls) as ctx:
+                be._call(f"judge this response: {placeholder}")
+            self.assertEqual(len(ctx.exception.pending), 1)
+
+    def test_reflect_retry_of_pending_reply_raises(self):
+        with tempfile.TemporaryDirectory() as hdir:
+            be = HandoffBackend(handoff_dir=hdir)
+            be._call("reflect on failures")
+            with self.assertRaises(PendingCalls):
+                be._call("reflect on failures\n\nIMPORTANT: "
+                         "your previous reply was not valid JSON. Reply with "
+                         "ONLY the JSON array, no prose, no markdown fences.")
+
+    def test_flush_writes_prompts_and_pending_json(self):
+        with tempfile.TemporaryDirectory() as hdir:
+            be = HandoffBackend(handoff_dir=hdir)
+            be._call("prompt A")
+            be._call("prompt B")
+            prompts_path = be.flush_pending()
+            self.assertTrue(os.path.exists(prompts_path))
+            with open(os.path.join(hdir, "pending.json"), encoding="utf-8") as f:
+                payload = json.load(f)
+            self.assertEqual(payload["format"], "skillopt_sleep.handoff.v1")
+            self.assertEqual(len(payload["pending"]), 2)
+            self.assertEqual(payload["pending"][0]["prompt"], "prompt A")
+            with open(prompts_path, encoding="utf-8") as f:
+                md = f.read()
+            self.assertIn("BEGIN PROMPT", md)
+            self.assertIn("prompt B", md)
+
+    def test_get_backend_registration(self):
+        with tempfile.TemporaryDirectory() as proj:
+            be = get_backend("handoff", project_dir=proj)
+            self.assertIsInstance(be, HandoffBackend)
+            self.assertTrue(be.handoff_dir.startswith(os.path.realpath(proj))
+                            or be.handoff_dir.startswith(proj))
+
+
+class TestHandoffCycle(unittest.TestCase):
+    def test_cycle_converges_over_handoff_rounds(self):
+        with tempfile.TemporaryDirectory() as proj, \
+                tempfile.TemporaryDirectory() as home:
+            hdir = os.path.join(proj, ".skillopt-sleep-handoff")
+
+            def cfg():
+                return load_config(
+                    invoked_project=proj, projects="invoked",
+                    backend="handoff",
+                    claude_home=os.path.join(home, ".claude"),
+                )
+
+            final = None
+            rounds = 0
+            for _ in range(8):
+                rounds += 1
+                backend = HandoffBackend(handoff_dir=hdir)
+                try:
+                    outcome = run_sleep_cycle(
+                        cfg(), seed_tasks=_tasks(), dry_run=True, backend=backend,
+                    )
+                except PendingCalls:
+                    outcome = None
+                if backend.pending:
+                    backend.flush_pending()
+                    _answer_pending(backend)
+                    continue
+                final = outcome
+                break
+
+            self.assertIsNotNone(final, "cycle never completed within 8 rounds")
+            self.assertGreater(rounds, 1, "expected at least one handoff round")
+            rep = final.report
+            self.assertTrue(rep.accepted, f"gate did not accept: {rep.gate_action}")
+            self.assertGreater(rep.candidate_score, rep.baseline_score)
+            self.assertTrue(any(RULE in e.content for e in rep.edits))
+
+
+class TestHandoffCli(unittest.TestCase):
+    def test_run_exits_3_and_stages_prompts(self):
+        from skillopt_sleep.__main__ import main
+        from skillopt_sleep.tasks_file import make_tasks_payload, write_tasks_file
+
+        with tempfile.TemporaryDirectory() as proj, \
+                tempfile.TemporaryDirectory() as home:
+            tasks_path = os.path.join(proj, "tasks.json")
+            payload = make_tasks_payload(_tasks(), project=proj)
+            payload["reviewed"] = True
+            write_tasks_file(tasks_path, payload)
+
+            rc = main([
+                "run", "--backend", "handoff",
+                "--project", proj,
+                "--claude-home", os.path.join(home, ".claude"),
+                "--tasks-file", tasks_path,
+            ])
+            self.assertEqual(rc, 3)
+            hdir = os.path.join(proj, ".skillopt-sleep-handoff")
+            self.assertTrue(os.path.exists(os.path.join(hdir, "PROMPTS.md")))
+            self.assertTrue(os.path.exists(os.path.join(hdir, "pending.json")))
+
+    def test_corrupt_digests_pin_falls_back_to_reharvest(self):
+        from skillopt_sleep.__main__ import main
+
+        with tempfile.TemporaryDirectory() as proj, \
+                tempfile.TemporaryDirectory() as home:
+            hdir = os.path.join(proj, ".skillopt-sleep-handoff")
+            os.makedirs(hdir, exist_ok=True)
+            with open(os.path.join(hdir, "digests.json"), "w", encoding="utf-8") as f:
+                f.write("{not valid json")
+            rc = main([
+                "run", "--backend", "handoff",
+                "--project", proj,
+                "--claude-home", os.path.join(home, ".claude"),
+            ])
+            # must not crash: corrupt pin -> fresh harvest -> no tasks -> 0
+            self.assertEqual(rc, 0)
+
+    def test_run_with_no_tasks_exits_0_and_advances_harvest_window(self):
+        from skillopt_sleep.__main__ import main
+        from skillopt_sleep.config import load_config
+        from skillopt_sleep.state import SleepState
+
+        with tempfile.TemporaryDirectory() as proj, \
+                tempfile.TemporaryDirectory() as home:
+            claude_home = os.path.join(home, ".claude")
+            rc = main([
+                "run", "--backend", "handoff",
+                "--project", proj,
+                "--claude-home", claude_home,
+            ])
+            self.assertEqual(rc, 0)
+            # the no-tasks branch must persist last-harvest, otherwise every
+            # later run re-scans the same stale window forever
+            cfg = load_config(invoked_project=proj, claude_home=claude_home)
+            state = SleepState.load(cfg.state_path)
+            self.assertIsNotNone(state.last_harvest_for(proj))
+
+    def test_no_tasks_removes_digests_pin_so_next_run_reharvests(self):
+        """After a no-task night the (often empty) digests.json pin must be
+        removed, or every later run reloads a []-pin and never harvests again."""
+        from skillopt_sleep.__main__ import main
+
+        with tempfile.TemporaryDirectory() as proj, \
+                tempfile.TemporaryDirectory() as home:
+            hdir = os.path.join(proj, ".skillopt-sleep-handoff")
+            os.makedirs(hdir, exist_ok=True)
+            digests_path = os.path.join(hdir, "digests.json")
+            with open(digests_path, "w", encoding="utf-8") as f:
+                json.dump([], f)
+            rc = main([
+                "run", "--backend", "handoff",
+                "--project", proj,
+                "--claude-home", os.path.join(home, ".claude"),
+            ])
+            self.assertEqual(rc, 0)
+            # the empty pin must be gone so the next run re-harvests
+            self.assertFalse(os.path.exists(digests_path))
+
+    def test_corrupt_tasks_pin_falls_back_to_remine(self):
+        """An interrupted/corrupt tasks.json snapshot must not crash the run."""
+        from skillopt_sleep.__main__ import main
+
+        with tempfile.TemporaryDirectory() as proj, \
+                tempfile.TemporaryDirectory() as home:
+            hdir = os.path.join(proj, ".skillopt-sleep-handoff")
+            os.makedirs(hdir, exist_ok=True)
+            with open(os.path.join(hdir, "tasks.json"), "w", encoding="utf-8") as f:
+                f.write('{"tasks": [ truncated...')
+            rc = main([
+                "run", "--backend", "handoff",
+                "--project", proj,
+                "--claude-home", os.path.join(home, ".claude"),
+            ])
+            # corrupt snapshot -> re-mine -> no tasks -> 0 (not a crash)
+            self.assertEqual(rc, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_qwen_backend.py

@@ -1,4 +1,5 @@
 """Tests for the OpenAI-compatible Qwen chat backend."""
+
 from __future__ import annotations
 
 import importlib.util
@@ -14,7 +15,6 @@ import pytest
 
 from skillopt.envs.searchqa.evaluator import extract_answer
 
-
 _QWEN_CONFIG_ENV_KEYS = (
     "BASE_URL",
     "API_KEY",
@@ -22,11 +22,10 @@ _QWEN_CONFIG_ENV_KEYS = (
     "TIMEOUT_SECONDS",
     "MAX_TOKENS",
     "ENABLE_THINKING",
+    "USE_MAX_COMPLETION_TOKENS",
 )
 _ENV_KEYS = ("OPTIMIZER_BACKEND", "TARGET_BACKEND") + tuple(
-    f"{prefix}QWEN_CHAT_{key}"
-    for prefix in ("", "OPTIMIZER_", "TARGET_")
-    for key in _QWEN_CONFIG_ENV_KEYS
+    f"{prefix}QWEN_CHAT_{key}" for prefix in ("", "OPTIMIZER_", "TARGET_") for key in _QWEN_CONFIG_ENV_KEYS
 )
 
 
@@ -225,3 +224,75 @@ def test_configure_qwen_chat_runtime_toggle_controls_payload(
 
     assert recorder.calls[0]["payload"]["chat_template_kwargs"] == {"enable_thinking": True}
     assert "chat_template_kwargs" not in recorder.calls[1]["payload"]
+
+
+def test_chat_target_uses_max_tokens_by_default(
+    monkeypatch: pytest.MonkeyPatch,
+    isolate_qwen_state: tuple[Any, Any],
+) -> None:
+    model_module, qwen_backend = isolate_qwen_state
+    _use_qwen_target(model_module, qwen_backend, enable_thinking=False)
+    recorder = _record_urlopen(monkeypatch, qwen_backend)
+
+    model_module.chat_target("system", "user", max_completion_tokens=128, retries=1)
+
+    payload = recorder.calls[0]["payload"]
+    assert payload["max_tokens"] == 128
+    assert "max_completion_tokens" not in payload
+
+
+def test_chat_target_uses_max_completion_tokens_when_enabled(
+    monkeypatch: pytest.MonkeyPatch,
+    isolate_qwen_state: tuple[Any, Any],
+) -> None:
+    model_module, qwen_backend = isolate_qwen_state
+    _use_qwen_target(model_module, qwen_backend, enable_thinking=False)
+    qwen_backend.TARGET_CONFIG.use_max_completion_tokens = True
+    recorder = _record_urlopen(monkeypatch, qwen_backend)
+
+    model_module.chat_target("system", "user", max_completion_tokens=128, retries=1)
+
+    payload = recorder.calls[0]["payload"]
+    assert payload["max_completion_tokens"] == 128
+    assert "max_tokens" not in payload
+
+
+def test_configure_qwen_chat_toggles_max_completion_tokens(
+    monkeypatch: pytest.MonkeyPatch,
+    isolate_qwen_state: tuple[Any, Any],
+) -> None:
+    model_module, qwen_backend = isolate_qwen_state
+    _use_qwen_target(model_module, qwen_backend, enable_thinking=False)
+    recorder = _record_urlopen(monkeypatch, qwen_backend)
+
+    model_module.configure_qwen_chat(target_use_max_completion_tokens=True)
+    model_module.chat_target("system", "user", max_completion_tokens=128, retries=1)
+    model_module.configure_qwen_chat(target_use_max_completion_tokens=False)
+    model_module.chat_target("system", "user", max_completion_tokens=128, retries=1)
+
+    assert "max_completion_tokens" in recorder.calls[0]["payload"]
+    assert "max_tokens" in recorder.calls[1]["payload"]
+
+
+def test_temperature_omitted_when_env_is_blank(
+    monkeypatch: pytest.MonkeyPatch,
+    isolate_qwen_state: tuple[Any, Any],
+) -> None:
+    _model_module, qwen_backend = isolate_qwen_state
+    # Explicit blank means "omit", not "fall back to 0.7".
+    monkeypatch.setenv("TARGET_QWEN_CHAT_TEMPERATURE", "")
+    assert qwen_backend._resolve_temperature("target") is None
+    monkeypatch.setenv("TARGET_QWEN_CHAT_TEMPERATURE", "off")
+    assert qwen_backend._resolve_temperature("target") is None
+
+
+def test_temperature_resolves_float_and_default(
+    monkeypatch: pytest.MonkeyPatch,
+    isolate_qwen_state: tuple[Any, Any],
+) -> None:
+    _model_module, qwen_backend = isolate_qwen_state
+    monkeypatch.delenv("QWEN_CHAT_TEMPERATURE", raising=False)
+    monkeypatch.delenv("TARGET_QWEN_CHAT_TEMPERATURE", raising=False)
+    assert qwen_backend._resolve_temperature("target") == 0.7
+    monkeypatch.setenv("TARGET_QWEN_CHAT_TEMPERATURE", "0.2")
+    assert qwen_backend._resolve_temperature("target") == 0.2

tests/test_semantic_density.py

@@ -0,0 +1,124 @@
+"""Tests for semantic density heuristic in the validation gate."""
+from __future__ import annotations
+
+import unittest
+from skillopt.evaluation.gate import (
+    compute_semantic_density,
+    select_gate_score,
+    evaluate_gate,
+)
+
+
+class TestSemanticDensity(unittest.TestCase):
+    """Test suite for semantic density scoring and gating decisions."""
+
+    def test_compute_semantic_density_basic(self) -> None:
+        """Verify basic compute_semantic_density behaviour with default words."""
+        # 10 words, 2 leading words ("always", "never") -> 0.2 density
+        skill = "Always check the inputs and never mix up proxy values."
+        density = compute_semantic_density(skill)
+        self.assertAlmostEqual(density, 0.2)
+
+        # Empty skill should have 0 density
+        self.assertEqual(compute_semantic_density(""), 0.0)
+        self.assertEqual(compute_semantic_density("   \n  "), 0.0)
+
+    def test_compute_semantic_density_custom_leading_words(self) -> None:
+        """Verify compute_semantic_density with custom leading words."""
+        skill = "Check the inputs carefully and resolve the equation."
+        leading = ["check", "resolve"]
+        # 8 words, 2 custom leading words -> 0.25 density
+        density = compute_semantic_density(skill, leading_words=leading)
+        self.assertAlmostEqual(density, 0.25)
+
+    def test_compute_semantic_density_with_protected_regions(self) -> None:
+        """Verify protected comments are excluded from density calculation."""
+        skill = (
+            "Always check inputs.\n"
+            "<!-- SLOW_UPDATE_START -->\n"
+            "This contains many words that should not count towards density "
+            "always and never and only.\n"
+            "<!-- SLOW_UPDATE_END -->\n"
+            "<!-- APPENDIX_START -->\n"
+            "More excluded words.\n"
+            "<!-- APPENDIX_END -->\n"
+        )
+        # Without stripping, there would be many more words and a different density.
+        # Stripped text: "Always check inputs." -> 3 words, 1 leading word ("always") -> 1/3 density
+        density = compute_semantic_density(skill)
+        self.assertAlmostEqual(density, 1.0 / 3.0)
+
+    def test_select_gate_score_no_density(self) -> None:
+        """Verify select_gate_score without semantic density adjustment."""
+        # Default behavior: no semantic density adjustment
+        score_hard = select_gate_score(0.8, 0.6, metric="hard")
+        self.assertEqual(score_hard, 0.8)
+
+        score_soft = select_gate_score(0.8, 0.6, metric="soft")
+        self.assertEqual(score_soft, 0.6)
+
+        score_mixed = select_gate_score(0.8, 0.6, metric="mixed", mixed_weight=0.5)
+        self.assertAlmostEqual(score_mixed, 0.7)
+
+    def test_select_gate_score_with_density(self) -> None:
+        """Verify select_gate_score with semantic density adjustment."""
+        # 10 words, 2 leading words ("always", "never") -> 0.2 density
+        skill = "Always check the inputs and never mix up proxy values."
+        # bonus: 0.1 (weight) * 0.2 (density) = 0.02
+        score = select_gate_score(
+            hard=0.8,
+            soft=0.6,
+            metric="hard",
+            skill_content=skill,
+            use_semantic_density=True,
+            semantic_density_weight=0.1,
+        )
+        self.assertAlmostEqual(score, 0.82)
+
+    def test_evaluate_gate_with_density_preference(self) -> None:
+        """Verify evaluate_gate prefers candidates with higher semantic density."""
+        # Baseline/current skill:
+        # "Always do this task step by step and be very careful because errors are bad."
+        # 15 words, 1 leading ("always") -> 1/15 density = ~0.0667
+        current_skill = "Always do this task step by step and be very careful because errors are bad."
+
+        # Candidate skill (shorter/more steerable):
+        # "Always verify outputs. Never mix proxy values."
+        # 7 words, 3 leading ("always", "verify", "never") -> 3/7 density = ~0.4286
+        candidate_skill = "Always verify outputs. Never mix proxy values."
+
+        # Both have same rollout accuracy (hard=0.8, soft=0.8)
+        # Baseline/current score: 0.8 + 0.1 * (1/15) = ~0.8067
+        current_score = select_gate_score(
+            hard=0.8,
+            soft=0.8,
+            metric="hard",
+            skill_content=current_skill,
+            use_semantic_density=True,
+            semantic_density_weight=0.1,
+        )
+
+        # Candidate score: 0.8 + 0.1 * (3/7) = ~0.8429
+        # Even though accuracy is equal, the candidate should be accepted due to higher semantic density
+        res = evaluate_gate(
+            candidate_skill=candidate_skill,
+            cand_hard=0.8,
+            current_skill=current_skill,
+            current_score=current_score,
+            best_skill=current_skill,
+            best_score=current_score,
+            best_step=1,
+            global_step=2,
+            cand_soft=0.8,
+            metric="hard",
+            use_semantic_density=True,
+            semantic_density_weight=0.1,
+        )
+
+        self.assertEqual(res.action, "accept_new_best")
+        self.assertEqual(res.current_skill, candidate_skill)
+        self.assertAlmostEqual(res.current_score, 0.8 + 0.1 * (3.0 / 7.0))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_sleep_engine.py

@@ -211,7 +211,7 @@ class TestHarvest(unittest.TestCase):
             self.assertTrue(cfg.get("progress"))
             self.assertEqual(
                 cfg.managed_skill_path(),
-                os.path.join(project, ".agents/skills/taste-skill/SKILL.md"),
+                os.path.abspath(os.path.join(project, ".agents/skills/taste-skill/SKILL.md")),
             )
 
     def test_cli_report_payload_includes_rejected_edits(self):
@@ -281,7 +281,7 @@ class TestHarvest(unittest.TestCase):
 
         self.assertEqual(
             cfg.managed_skill_path(),
-            "/repo/Yoshi/.agents/skills/yoshi-monorepo/SKILL.md",
+            os.path.abspath("/repo/Yoshi/.agents/skills/yoshi-monorepo/SKILL.md"),
         )
 
     def test_cmd_run_uses_tasks_file_without_harvest(self):
@@ -912,7 +912,7 @@ class TestFullCycleAndAdopt(unittest.TestCase):
 
     def test_cycle_can_target_repo_scoped_skill_path(self):
         with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
-            target = os.path.join(proj, ".agents/skills/taste-skill/SKILL.md")
+            target = os.path.abspath(os.path.join(proj, ".agents/skills/taste-skill/SKILL.md"))
             cfg = load_config(
                 invoked_project=proj,
                 projects="invoked",
@@ -1118,6 +1118,36 @@ class TestClaudeCliBackendBare(unittest.TestCase):
         # But it's also recorded for detection
         self.assertIn("Not logged in", getattr(be, "last_call_error", ""))
 
+    def test_spawn_failure_sets_last_call_error(self):
+        """When subprocess.run raises FileNotFoundError, _call must set
+        last_call_error and log a warning instead of silently returning ''."""
+        from skillopt_sleep.backend import ClaudeCliBackend
+        be = ClaudeCliBackend(
+            claude_path="/nonexistent/claude-binary",
+            timeout=3,
+        )
+        result = be._call("test prompt")
+        self.assertEqual(result, "")
+        self.assertIn("Claude CLI spawn failed", be.last_call_error)
+
+    def test_attempt_tools_spawn_failure_sets_last_call_error(self):
+        """When subprocess.run raises in attempt_with_tools, last_call_error
+        must be set and a warning logged."""
+        from skillopt_sleep.backend import ClaudeCliBackend
+        from skillopt_sleep.types import TaskRecord
+        be = ClaudeCliBackend(
+            claude_path="/nonexistent/claude-binary",
+            timeout=3,
+        )
+        task = TaskRecord(
+            id="t1", project="/p", intent="test",
+            reference="ref", reference_kind="exact",
+            tags=[], split="train",
+        )
+        resp, called = be.attempt_with_tools(task, "", "", tools=["search"])
+        self.assertEqual(resp, "")
+        self.assertIn("Claude CLI spawn failed", be.last_call_error)
+
 
 
 
@@ -1147,6 +1177,32 @@ class MockRewardHackingBackend(MockBackend):
                 break
         return edits
 
+class MockBeneficialBackend(MockBackend):
+    def attempt(self, task: TaskRecord, skill: str, memory: str, sample_id: int = 0) -> str:
+        ctx = (skill or "") + "\n" + (memory or "")
+        # If the beneficial artifact is present:
+        if "Always think step-by-step" in ctx:
+            if "rule:__beneficial__" in task.tags:
+                return str(task.reference) # improves the train task
+            if "rule:real" in task.tags:
+                return str(task.reference) # improves the real held-out task
+                
+        # Baseline behavior (without the rule)
+        if "rule:__beneficial__" in task.tags:
+            return "I am missing input" # baseline fails the train task
+        if "rule:real" in task.tags:
+            return "baseline fails too" # baseline fails the real task
+            
+        return super().attempt(task, skill, memory, sample_id)
+
+    def reflect(self, failures, successes, skill, memory, **kwargs):
+        edits = []
+        for task, _ in failures:
+            if "rule:__beneficial__" in task.tags:
+                edits.append(EditRecord("skill", "add", "Always think step-by-step", rationale="beneficial"))
+                break
+        return edits
+
 class TestVerifierDiscipline(unittest.TestCase):
     def test_gate_rejects_reward_hacking_edit(self):
         be = MockRewardHackingBackend()
@@ -1157,9 +1213,27 @@ class TestVerifierDiscipline(unittest.TestCase):
         res = consolidate(be, tasks, "", "", edit_budget=4, gate_metric="hard", night=1)
         
         self.assertFalse(res.accepted)
+        self.assertEqual(res.gate_action, "reject")
+        self.assertEqual(res.holdout_baseline, 1.0)
+        self.assertEqual(res.holdout_candidate, 1.0) # final state reverts to baseline
         self.assertGreater(len(res.rejected_edits), 0)
         self.assertIn("placeholder", res.rejected_edits[0].content)
 
+    def test_gate_accepts_beneficial_edit(self):
+        be = MockBeneficialBackend()
+        train_task = TaskRecord(id="t2", project="/p", intent="train", reference="ABCDEF", reference_kind="exact", tags=["rule:__beneficial__"], split="train")
+        val_task = TaskRecord(id="v2", project="/p", intent="val", reference="UVWXYZ", reference_kind="exact", tags=["rule:real"], split="val")
+        tasks = [train_task, val_task]
+
+        res = consolidate(be, tasks, "", "", edit_budget=4, gate_metric="hard", night=1)
+        
+        self.assertTrue(res.accepted)
+        self.assertEqual(res.gate_action, "accept_new_best")
+        self.assertEqual(res.holdout_baseline, 0.0)
+        self.assertEqual(res.holdout_candidate, 1.0)
+        self.assertGreater(len(res.applied_edits), 0)
+        self.assertIn("step-by-step", res.applied_edits[0].content)
+
 class TestDiagnosticsRedaction(unittest.TestCase):
     """diagnostics.json surfaces backend stderr / optimizer replies / task
     responses for debugging — but those can carry credentials (e.g. a codex 401