feat(sleep): add handoff backend — session-executed model calls, no API subprocess (#125)
Adds --backend handoff: the engine runs all deterministic stages and outsources attempt/judge/reflect to prompt/answer files an interactive agent session fills between runs (exit 3 = pending batch, re-run to resume). Deterministic replay + the prompt-hash answer cache make resume stateless; sentinel detection aborts any call built from unanswered output so placeholders never reach scores or staging. Session digests and mined tasks are pinned per night (secret-redacted) so the sessions answering prompts cannot shift the task set, and LLM mining is routed through the same handoff files. Ships a /skillopt-sleep-handoff Claude Code command that answers each prompt in a fresh-context subagent to protect the held-out gate. Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
dimitarvdenev
+191
-2
@@ -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,187 @@ 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()
|
||||
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")
|
||||
if os.path.exists(snapshot):
|
||||
tasks, _meta = load_tasks_file(
|
||||
snapshot,
|
||||
holdout_fraction=cfg.get("holdout_fraction", 0.34),
|
||||
seed=cfg.get("seed", 42),
|
||||
)
|
||||
else:
|
||||
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
|
||||
|
||||
|
||||
|
||||
@@ -1415,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()
|
||||
|
||||
|
||||
|
||||
@@ -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", ""),
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user