mirror of
https://github.com/microsoft/SkillOpt.git
synced 2026-07-05 23:30:35 +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>
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""SkillOpt-Sleep — persistent cross-night state.
|
|
|
|
state.json lives in ~/.skillopt-sleep and is the "long-term" store that
|
|
turns nightly episodes into durable competence (the Agent-Sleep paper's
|
|
short-term -> long-term transfer). It records:
|
|
|
|
- night counter
|
|
- last harvest timestamp per project (so each night only sees new data)
|
|
- cross-night "slow/meta" memory (lessons that persisted across nights)
|
|
- per-night history (scores, accept/reject) for trend reporting
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
def _now_iso(clock: Optional[float] = None) -> str:
|
|
# caller passes a timestamp; we avoid importing time at module import
|
|
import time as _t
|
|
return _t.strftime("%Y-%m-%dT%H:%M:%S", _t.localtime(clock if clock is not None else _t.time()))
|
|
|
|
|
|
DEFAULT_STATE: Dict[str, Any] = {
|
|
"version": 1,
|
|
"night": 0,
|
|
"last_harvest": {}, # project -> iso timestamp of last harvested record
|
|
"slow_memory": "", # cross-night consolidated lessons (meta-skill analogue)
|
|
"history": [], # list of per-night summaries
|
|
}
|
|
|
|
|
|
class SleepState:
|
|
def __init__(self, path: str, data: Optional[Dict[str, Any]] = None) -> None:
|
|
self.path = path
|
|
self.data = data if data is not None else dict(DEFAULT_STATE)
|
|
|
|
# io ---------------------------------------------------------------------
|
|
@classmethod
|
|
def load(cls, path: str) -> "SleepState":
|
|
if os.path.exists(path):
|
|
try:
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
merged = dict(DEFAULT_STATE)
|
|
merged.update(data if isinstance(data, dict) else {})
|
|
return cls(path, merged)
|
|
except Exception:
|
|
pass
|
|
return cls(path, dict(DEFAULT_STATE))
|
|
|
|
def save(self) -> None:
|
|
os.makedirs(os.path.dirname(self.path), exist_ok=True)
|
|
tmp = self.path + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(self.data, f, ensure_ascii=False, indent=2)
|
|
os.replace(tmp, self.path)
|
|
|
|
# accessors --------------------------------------------------------------
|
|
@property
|
|
def night(self) -> int:
|
|
return int(self.data.get("night", 0))
|
|
|
|
def last_harvest_for(self, project: str) -> Optional[str]:
|
|
return self.data.get("last_harvest", {}).get(project)
|
|
|
|
def set_last_harvest(self, project: str, iso_ts: str) -> None:
|
|
self.data.setdefault("last_harvest", {})[project] = iso_ts
|
|
|
|
@property
|
|
def slow_memory(self) -> str:
|
|
return str(self.data.get("slow_memory", ""))
|
|
|
|
def set_slow_memory(self, content: str) -> None:
|
|
self.data["slow_memory"] = content
|
|
|
|
def begin_night(self, clock: Optional[float] = None) -> int:
|
|
self.data["night"] = self.night + 1
|
|
return self.night
|
|
|
|
def record_night(self, summary: Dict[str, Any]) -> None:
|
|
self.data.setdefault("history", []).append(summary)
|