mirror of
https://github.com/microsoft/SkillOpt.git
synced 2026-07-08 00:49:57 +08:00
b02ffc2c99
Open-source-tool / research-code separation:
- git mv skillopt/sleep/ -> skillopt_sleep/ (top-level, sibling to the research
skillopt/ package). History preserved as renames.
- All imports skillopt.sleep.* -> skillopt_sleep.*.
- Vendor the validation gate into skillopt_sleep/gate.py (a self-contained copy
of skillopt.evaluation.gate). The engine now has ZERO dependency on the
research package — verified: grep finds no `from skillopt.` in skillopt_sleep/,
and consolidate's gate resolves to skillopt_sleep.gate.
- Plugin scripts/commands/skill call `-m skillopt_sleep`.
29 tests pass; `python -m skillopt_sleep` runs standalone.
Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
"""SkillOpt-Sleep — Stage 5/6: staging and adoption.
|
|
|
|
Implements the Dreams safety contract: the cycle never mutates the user's
|
|
live CLAUDE.md / SKILL.md. It writes proposals + a human-readable report into
|
|
a staging directory; a separate, explicit `adopt` step copies them over the
|
|
live files after taking a backup.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import time
|
|
from typing import List, Optional
|
|
|
|
from skillopt_sleep.types import SleepReport
|
|
|
|
|
|
def _ts_dir() -> str:
|
|
return time.strftime("%Y%m%d-%H%M%S", time.localtime())
|
|
|
|
|
|
def staging_root(project: str) -> str:
|
|
return os.path.join(project, ".skillopt-sleep", "staging")
|
|
|
|
|
|
def latest_staging(project: str) -> Optional[str]:
|
|
root = staging_root(project)
|
|
if not os.path.isdir(root):
|
|
return None
|
|
subs = sorted(
|
|
(os.path.join(root, d) for d in os.listdir(root)),
|
|
key=lambda p: os.path.getmtime(p),
|
|
reverse=True,
|
|
)
|
|
return subs[0] if subs else None
|
|
|
|
|
|
def write_staging(
|
|
project: str,
|
|
*,
|
|
report: SleepReport,
|
|
proposed_skill: Optional[str],
|
|
proposed_memory: Optional[str],
|
|
live_skill_path: str,
|
|
live_memory_path: str,
|
|
report_md: str,
|
|
) -> str:
|
|
"""Write proposals + report into staging/<ts>/ and return that path."""
|
|
out = os.path.join(staging_root(project), _ts_dir())
|
|
os.makedirs(out, exist_ok=True)
|
|
|
|
manifest = {
|
|
"live_skill_path": live_skill_path,
|
|
"live_memory_path": live_memory_path,
|
|
"has_skill": proposed_skill is not None,
|
|
"has_memory": proposed_memory is not None,
|
|
"accepted": report.accepted,
|
|
}
|
|
if proposed_skill is not None:
|
|
with open(os.path.join(out, "proposed_SKILL.md"), "w", encoding="utf-8") as f:
|
|
f.write(proposed_skill)
|
|
if proposed_memory is not None:
|
|
with open(os.path.join(out, "proposed_CLAUDE.md"), "w", encoding="utf-8") as f:
|
|
f.write(proposed_memory)
|
|
with open(os.path.join(out, "report.json"), "w", encoding="utf-8") as f:
|
|
json.dump(report.to_dict(), f, ensure_ascii=False, indent=2)
|
|
with open(os.path.join(out, "report.md"), "w", encoding="utf-8") as f:
|
|
f.write(report_md)
|
|
with open(os.path.join(out, "manifest.json"), "w", encoding="utf-8") as f:
|
|
json.dump(manifest, f, ensure_ascii=False, indent=2)
|
|
return out
|
|
|
|
|
|
def _backup(path: str, backup_dir: str) -> None:
|
|
if os.path.exists(path):
|
|
os.makedirs(backup_dir, exist_ok=True)
|
|
shutil.copy2(path, os.path.join(backup_dir, os.path.basename(path)))
|
|
|
|
|
|
def adopt(staging_dir: str) -> List[str]:
|
|
"""Copy staged proposals over the live files, backing up first.
|
|
|
|
Returns the list of live paths that were updated.
|
|
"""
|
|
with open(os.path.join(staging_dir, "manifest.json")) as f:
|
|
manifest = json.load(f)
|
|
backup_dir = os.path.join(staging_dir, "backup")
|
|
updated: List[str] = []
|
|
|
|
if manifest.get("has_skill"):
|
|
live = manifest["live_skill_path"]
|
|
os.makedirs(os.path.dirname(live), exist_ok=True)
|
|
_backup(live, backup_dir)
|
|
shutil.copy2(os.path.join(staging_dir, "proposed_SKILL.md"), live)
|
|
updated.append(live)
|
|
if manifest.get("has_memory"):
|
|
live = manifest["live_memory_path"]
|
|
os.makedirs(os.path.dirname(live), exist_ok=True)
|
|
_backup(live, backup_dir)
|
|
shutil.copy2(os.path.join(staging_dir, "proposed_CLAUDE.md"), live)
|
|
updated.append(live)
|
|
return updated
|