Yifan Yang
4e7add899d
Add skillopt/sleep — a deployment-time companion to SkillOpt that gives a
local Claude agent a nightly "sleep cycle":
harvest ~/.claude transcripts -> mine recurring tasks -> replay offline
-> consolidate (reflect -> bounded edit -> held-out GATE) -> stage -> adopt
Synthesizes SkillOpt (validation-gated bounded text optimization, reusing
skillopt.evaluation.gate verbatim), Claude Dreams (offline consolidation;
input never mutated; review-then-adopt), and the agent-sleep paper
(short-term experience -> long-term competence).
Engine (skillopt/sleep/, import-light, py>=3.10):
- harvest.py read-only parse of session JSONL + history.jsonl
- mine.py sessions -> TaskRecords (heuristic miner + LLM hook)
- backend.py MockBackend (deterministic, no API) + AnthropicBackend
- replay.py offline re-run -> (hard, soft) scores
- consolidate.py one SkillOpt epoch behind a held-out gate
- memory.py protected-region edits to SKILL.md / CLAUDE.md
- staging.py stage proposals; adopt with backup (Dreams safety contract)
- cycle.py + __main__.py orchestrator + CLI (run/dry-run/status/adopt/harvest)
Plugin (skillopt-sleep-plugin/): plugin.json, /sleep command, skillopt-sleep
skill, SessionEnd hook, bundled runner + cron generator.
Validation (deterministic, no API): persona experiment proves held-out lift
(researcher 0.33->1.0, programmer 0.32->1.0) AND that the gate rejects an
injected harmful edit. 13 stdlib-unittest tests pass, incl. full cycle +
adopt-with-backup and parsing of real on-disk transcripts.
Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
"""SkillOpt-Sleep — Stage 3: replay.
|
|
|
|
Re-run mined TaskRecords offline under a given (skill, memory) and score
|
|
them, producing the (hard, soft) signal SkillOpt's gate consumes.
|
|
|
|
For Phase 1 the replay is "mock mode": a sandboxed single-shot attempt via
|
|
the chosen backend (MockBackend = deterministic; AnthropicBackend = real).
|
|
"fresh" worktree replay is Phase 3 and is intentionally not wired here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Tuple
|
|
|
|
from skillopt.sleep.backend import Backend
|
|
from skillopt.sleep.types import ReplayResult, TaskRecord
|
|
|
|
|
|
def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str) -> ReplayResult:
|
|
response = backend.attempt(task, skill, memory)
|
|
hard, soft, rationale = backend.judge(task, response)
|
|
return ReplayResult(
|
|
id=task.id,
|
|
hard=float(hard),
|
|
soft=float(soft),
|
|
response=response,
|
|
fail_reason="" if hard >= 1.0 else (rationale or "below threshold"),
|
|
task_type=(task.tags[0] if task.tags else "task"),
|
|
judge_rationale=rationale,
|
|
)
|
|
|
|
|
|
def replay_batch(
|
|
backend: Backend,
|
|
tasks: List[TaskRecord],
|
|
skill: str,
|
|
memory: str,
|
|
) -> List[Tuple[TaskRecord, ReplayResult]]:
|
|
return [(t, replay_one(backend, t, skill, memory)) for t in tasks]
|
|
|
|
|
|
def aggregate_scores(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> Tuple[float, float]:
|
|
if not pairs:
|
|
return 0.0, 0.0
|
|
hard = sum(r.hard for _t, r in pairs) / len(pairs)
|
|
soft = sum(r.soft for _t, r in pairs) / len(pairs)
|
|
return hard, soft
|