github/microsoft-SkillOpt
1
0
Fork 0
mirror of https://github.com/microsoft/SkillOpt.git synced 2026-08-03 07:02:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(sleep): per-night evidence chain and live prompt registry

Browse Source 
* skillopt_sleep/evidence.py — append-only, thread-safe, redacted
  evidence.jsonl per night: harvest sessions -> miner exchanges (verbatim
  prompt/reply) -> mined tasks with checks -> split assignment -> every
  replay attempt (phase-tagged, cache hits marked) -> per-task scores with
  failing checks named -> reflect exchanges + parsed edits -> gate trials
  and the final decision with its score arithmetic -> staged artifacts.
  Config-gated (evidence_log, default on; evidence_max_chars cap).

* skillopt_sleep/prompts.py — central registry of the four LLM prompt
  templates (miner/attempt/judge/reflect), byte-identical defaults to the
  previously inlined strings; user overrides in prompts.json take effect
  on the next model call (mtime-checked), no restart needed.

* cycle.py builds dual backends from config (optimizer_*/target_*),
  pre-creates the staging dir so evidence lands beside the report;
  staging.new_staging_dir() de-collides same-second runs;
  latest_staging() now skips non-adoptable (evidence-only) folders.

* tests/test_sleep_evidence.py — 8 no-network stdlib tests: chain
  completeness, redaction/truncation/ordering, disable flag, prompt
  override round-trip + live effect, no-tasks-night adoption guard.
This commit is contained in:
Alpha
2026-07-18 21:51:46 -06:00
committed by Yif-Yang
parent b39081768a
commit 08cb5a9cb6
10 changed files with 789 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
skillopt_sleep/backend.py
View File

@@ -47,6 +47,11 @@ class Backend:
name = "base"
# Optional user preferences (free text) injected into reflect as a prior.
preferences: str = ""
# Optional per-night evidence log (skillopt_sleep.evidence.EvidenceLog).
# Attached by the cycle; None => no observability overhead. The phase tag
# labels which consolidation step subsequent replay calls belong to.
evidence = None
evidence_phase: str = ""
def attempt(self, task: TaskRecord, skill: str, memory: str,
sample_id: int = 0) -> str:
@@ -330,11 +335,23 @@ class CliBackend(Backend):
raise NotImplementedError
def _cached_call(self, key: str, prompt: str, *, max_tokens: int = 1024) -> str:
kind = key.split(":", 1)[0]
ev = getattr(self, "evidence", None)
if key in self._cache:
# cache hits log key-only (the full text is on the original miss event)
if ev is not None:
ev.log("replay", "model_call", kind=kind, cache_hit=True, key=key,
phase=getattr(self, "evidence_phase", ""), backend=self.name,
model=self.model)
return self._cache[key]
out = self._call(prompt, max_tokens=max_tokens)
self._tokens += len(prompt) // 4 + len(out) // 4
self._cache[key] = out
if ev is not None:
ev.log("replay", "model_call", kind=kind, cache_hit=False, key=key,
phase=getattr(self, "evidence_phase", ""), backend=self.name,
model=self.model, prompt=prompt, response=out,
error=getattr(self, "last_call_error", "") or "")
return out
# operations -----------------------------------------------------------
@@ -363,16 +380,15 @@ class CliBackend(Backend):
return self._cached_call(key, prompt, max_tokens=512)
# generic path (mined daily-case tasks): neutral, content-filter-safe
# wording. Apply the skill/memory as guidance, not as adversarial
# "OVERRIDE everything" directives.
prompt = (
"Complete the following task for the user. Follow the skill and memory "
"guidance below, including any output-format and length requirements. "
"When a 'Learned preferences' rule sets an explicit limit (e.g. a length "
"cap), prefer that rule over more general advice it refines.\n\n"
f"# Skill\n{skill or '(none)'}\n\n# Memory\n{memory or '(none)'}\n\n"
f"# Task\n{task.intent}\n\n{task.context_excerpt}\n\n"
"Return ONLY the final answer text, nothing else."
)
# "OVERRIDE everything" directives. Template lives in the prompt
# registry so the dashboard can display/override it live.
from skillopt_sleep import prompts as prompt_registry
prompt = prompt_registry.render("attempt", {
"__SKILL__": skill or "(none)",
"__MEMORY__": memory or "(none)",
"__INTENT__": task.intent,
"__CONTEXT__": task.context_excerpt,
})
# cache on (task, skill, memory) so identical hold-out re-scoring is free
salt = f"s{sample_id}:" if sample_id else ""
key = "attempt:" + salt + skill_hash(prompt)
@@ -395,11 +411,11 @@ class CliBackend(Backend):
if task.reference_kind == "exact" and task.reference:
hard = exact_score(task.reference, response)
return hard, max(hard, keyword_soft_score(task.reference, response)), "exact(local)"
prompt = (
"Score how well the response satisfies the rubric, 0..1. "
'Return ONLY JSON {"score": <0..1>, "reason": "..."}.\n\n'
f"# Rubric\n{task.reference or task.intent}\n\n# Response\n{response}"
)
from skillopt_sleep import prompts as prompt_registry
prompt = prompt_registry.render("judge", {
"__RUBRIC__": task.reference or task.intent,
"__RESPONSE__": response,
})
key = "judge:" + skill_hash(prompt)
raw = self._cached_call(key, prompt, max_tokens=200)
obj = _extract_json(raw, "object")
@@ -482,39 +498,20 @@ class CliBackend(Backend):
# can't ask questions). We surface the benchmark's own rollout system
# prompt (carried on TaskRecord.system) so proposed rules stay in-bounds.
guard_text = _task_guardrail(failures)
prompt = (
"You are SkillOpt's optimizer. The agent keeps failing the recurring "
f"tasks below. Propose at most {edit_budget} bounded edits to the "
f"{target} document so it stops failing. Each edit MUST be a short, "
"GENERAL, reusable rule or preference (never task-specific, never an "
"answer to a single task). If exact failing criteria are listed, your "
"edits MUST make future outputs satisfy every one of them.\n"
"BE CONCRETE: quote the exact threshold, section name, or format from "
"the criteria verbatim in your rule (e.g. write 'keep the entire "
"response under 1200 characters', NOT 'respect length limits'). Vague "
"rules do not change behavior; specific numeric/structural rules do.\n"
"IMPORTANT: your edits are APPENDED to a 'Learned preferences' block; "
"you CANNOT delete the existing instructions above. If the current "
f"{target} text conflicts with a criterion (e.g. it says 'be exhaustive' "
"but outputs must be under a character limit), write an explicit, "
"forceful OVERRIDE rule stating it supersedes the conflicting "
"instruction, and put the hard requirement first.\n"
"HARD CONSTRAINT: every rule you write MUST be consistent with the "
"'Task output contract' below (if shown). NEVER propose a rule that "
"changes the required output format/language, tells the agent to ask "
"the user a question, or otherwise violates that contract — such a "
"rule scores ZERO because the evaluator cannot honor it.\n"
'Return ONLY a JSON array: '
'[{"op":"add|replace|delete","content":"<rule>","anchor":"<text to replace/delete, optional>","rationale":"<why>"}].\n\n'
f"# Current {target}\n{cur_doc}\n"
f"{guard_text}"
f"{criteria_text}\n"
f"{pref_text}\n\n"
f"# Recurring failures\n{fail_text}"
)
from skillopt_sleep import prompts as prompt_registry
prompt = prompt_registry.render("reflect", {
"__EDIT_BUDGET__": str(edit_budget),
"__TARGET__": target,
"__CUR_DOC__": cur_doc,
"__GUARD__": guard_text,
"__CRITERIA__": criteria_text,
"__PREFS__": pref_text,
"__FAILURES__": fail_text,
})
# Call with one retry: transient non-JSON replies otherwise waste a whole
# night (the gate sees no edits and rejects). A firmer second prompt
# recovers most of these.
ev = getattr(self, "evidence", None)
arr = None
for attempt in range(2):
p = prompt if attempt == 0 else (
@@ -523,6 +520,11 @@ class CliBackend(Backend):
)
raw = self._call(p, max_tokens=1024)
self._tokens += len(p) // 4 + len(raw) // 4
if ev is not None:
ev.log("reflect", "exchange", target=target, attempt=attempt + 1,
backend=self.name, model=self.model,
n_failures=len(failures), prompt=p, raw_reply=raw,
error=getattr(self, "last_call_error", "") or "")
arr = _extract_json(raw, "array")
if isinstance(arr, list) and arr:
break

11
skillopt_sleep/config.py
View File

@@ -40,6 +40,14 @@ DEFAULTS: Dict[str, Any] = {
# ── optimizer ──────────────────────────────────────────────────────────
"backend": "mock", # "mock" | "claude" | "codex" | "copilot" | "cursor"
"model": "", # backend-specific; "" => backend default
# Dual-backend split (both empty => single backend above plays all roles).
# target = the model whose skill is deployed (runs `attempt` rollouts);
# optimizer = the model that mines tasks, judges rubrics, writes edits.
"optimizer_backend": "",
"optimizer_model": "",
"target_backend": "",
"target_model": "",
"azure_endpoint": "", # explicit endpoint for azure/compat backends
"gate_mode": "on", # "on" (validation-gated) | "off" (greedy, no hard filter)
"codex_path": "", # "" => auto-detect the real @openai/codex binary
"cursor_path": "", # "" => auto-detect the Cursor Agent CLI
@@ -58,6 +66,9 @@ DEFAULTS: Dict[str, Any] = {
"target_skill_path": "", # explicit SKILL.md target for repo-scoped agents
"target_task_filter": True, # prefer mined tasks matching target_skill_path/text
"progress": False, # print phase progress to stderr
# ── observability ──────────────────────────────────────────────────────
"evidence_log": True, # write per-night evidence.jsonl (full evidentiary chain)
"evidence_max_chars": 4000, # per-field truncation cap for evidence events
# ── adoption / safety ──────────────────────────────────────────────────
"auto_adopt": False, # default: stage + require explicit `adopt`
"managed_skill_name": "skillopt-sleep-learned",

51
skillopt_sleep/consolidate.py
View File

@@ -108,6 +108,8 @@ def consolidate(
Skill and memory are evolved in sequence (skill first if both enabled).
"""
from skillopt_sleep import evidence as evlog
ev = evlog.get(backend)
train_tasks, val_tasks = _split(tasks)
gate_off = str(gate_mode).strip().lower() in {"off", "none", "false", "greedy"}
holdout_detail: List[dict] = []
@@ -120,12 +122,19 @@ def consolidate(
if gate_off:
base_hard, base_soft = 0.0, 0.0
else:
evlog.set_phase(backend, "baseline_val")
base_pairs = replay_batch(backend, val_tasks, skill, memory)
base_hard, base_soft = aggregate_scores(base_pairs)
holdout_detail = _holdout_detail(base_pairs)
base_score = select_gate_score(base_hard, base_soft, gate_metric, gate_mixed_weight)
if ev is not None:
ev.log("gate", "baseline", gate_mode=("off" if gate_off else "on"),
n_train=len(train_tasks), n_val=len(val_tasks),
hard=base_hard, soft=base_soft, score=base_score,
metric=gate_metric, mixed_weight=gate_mixed_weight)
# ── reflect over TRAIN-split failures/successes ───────────────────────
evlog.set_phase(backend, "train")
train_pairs = replay_batch(backend, train_tasks, skill, memory)
failures = [(t, r) for (t, r) in train_pairs if r.hard < 1.0]
successes = [(t, r) for (t, r) in train_pairs if r.hard >= 1.0]
@@ -134,8 +143,15 @@ def consolidate(
all_applied: List[EditRecord] = []
all_rejected: List[EditRecord] = []
def _edits_payload(edits: List[EditRecord]) -> List[dict]:
return [{"op": e.op, "content": e.content, "anchor": e.anchor,
"rationale": e.rationale} for e in edits]
def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
nonlocal cand_skill, cand_memory, base_score, all_applied, all_rejected
if ev is not None:
ev.log("reflect", "edits_returned", target=which,
n_edits=len(edits), edits=_edits_payload(edits))
if not edits:
return doc
new_doc, applied = apply_edits(doc, edits)
@@ -144,14 +160,24 @@ def consolidate(
# gate OFF: accept greedily with NO val scoring (the daily-use path)
if gate_off:
all_applied.extend(applied)
if ev is not None:
ev.log("gate", "trial", target=which, mode="greedy",
accepted=True, n_edits=len(applied))
return new_doc
# gate ON: score the candidate on the VAL slice, keep only if it improves
trial_skill = new_doc if which == "skill" else cand_skill
trial_memory = new_doc if which == "memory" else cand_memory
evlog.set_phase(backend, f"gate_trial:{which}")
pairs = replay_batch(backend, val_tasks, trial_skill, trial_memory)
h, s = aggregate_scores(pairs)
cand_score = select_gate_score(h, s, gate_metric, gate_mixed_weight)
if cand_score > base_score:
improved = cand_score > base_score
if ev is not None:
ev.log("gate", "trial", target=which, mode="gated",
baseline_score=base_score, cand_hard=h, cand_soft=s,
cand_score=cand_score, accepted=improved,
n_edits=len(applied))
if improved:
base_score = max(base_score, cand_score)
all_applied.extend(applied)
return new_doc
@@ -204,6 +230,7 @@ def consolidate(
if evolve_memory:
# re-evaluate failures under the (possibly improved) skill
evlog.set_phase(backend, "train_post_skill")
train_pairs2 = replay_batch(backend, train_tasks, cand_skill, cand_memory)
failures2 = [(t, r) for (t, r) in train_pairs2 if r.hard < 1.0]
successes2 = [(t, r) for (t, r) in train_pairs2 if r.hard >= 1.0]
@@ -225,6 +252,7 @@ def consolidate(
base_gate_score = 0.0
else:
# scored on the VAL slice (the gate reference)
evlog.set_phase(backend, "final_val")
final_pairs = replay_batch(backend, val_tasks, cand_skill, cand_memory)
final_hard, final_soft = aggregate_scores(final_pairs)
final_score = select_gate_score(final_hard, final_soft, gate_metric, gate_mixed_weight)
@@ -249,6 +277,27 @@ def consolidate(
action = "accept" if final_score > base_gate_score else "reject"
accepted = bool(all_applied) and final_score > base_gate_score
if ev is not None:
w = max(0.0, min(1.0, float(gate_mixed_weight)))
if gate_metric == "mixed":
formula = (
f"score = (1-{w})*hard + {w}*soft; "
f"baseline = (1-{w})*{base_hard:.3f} + {w}*{base_soft:.3f} = {base_gate_score:.3f}; "
f"candidate = (1-{w})*{final_hard:.3f} + {w}*{final_soft:.3f} = {final_score:.3f}"
)
else:
formula = (
f"score = {gate_metric}; baseline = {base_gate_score:.3f}; "
f"candidate = {final_score:.3f}"
)
ev.log("gate", "decision", action=action, accepted=accepted,
baseline_score=base_gate_score, candidate_score=final_score,
baseline_hard=base_hard, baseline_soft=base_soft,
candidate_hard=final_hard, candidate_soft=final_soft,
metric=gate_metric, mixed_weight=gate_mixed_weight,
formula=formula, n_applied=len(all_applied),
n_rejected=len(all_rejected), night=night)
return ConsolidationResult(
accepted=accepted,
gate_action=action,

83
skillopt_sleep/cycle.py
View File

@@ -14,7 +14,9 @@ import sys
from dataclasses import dataclass
from typing import List, Optional
from skillopt_sleep.backend import Backend, get_backend
from skillopt_sleep import evidence
from skillopt_sleep.backend import Backend, build_backend
from skillopt_sleep.evidence import EvidenceLog
from skillopt_sleep.config import SleepConfig, load_config
from skillopt_sleep.dream import dream_consolidate
from skillopt_sleep.harvest_sources import harvest_for_config
@@ -114,16 +116,51 @@ def run_sleep_cycle(
project = _project_paths(cfg)
started = _now_iso(clock)
backend = backend or get_backend(
cfg.get("backend", "mock"),
backend = backend or build_backend(
backend=cfg.get("backend", "mock"),
model=cfg.get("model", ""),
optimizer_backend=cfg.get("optimizer_backend", ""),
optimizer_model=cfg.get("optimizer_model", ""),
target_backend=cfg.get("target_backend", ""),
target_model=cfg.get("target_model", ""),
codex_path=cfg.get("codex_path", ""),
cursor_path=cfg.get("cursor_path", ""),
azure_endpoint=cfg.get("azure_endpoint", ""),
preferences=cfg.get("preferences", ""),
project_dir=project,
)
backend.preferences = cfg.get("preferences", "")
_progress(cfg, f"night {night}: project={project} backend={backend.name}")
# ── evidence log (the night's full evidentiary chain) ────────────────
# Pre-create the staging dir so evidence.jsonl accumulates exactly where
# the report will land; dry-runs log into the state dir instead.
ev = None
staging_dir_pre = ""
if cfg.get("evidence_log", True):
from skillopt_sleep.staging import _ts_dir, new_staging_dir
if dry_run:
ev_path = os.path.join(
cfg.state_dir, "evidence", f"dryrun-{_ts_dir()}.jsonl")
else:
staging_dir_pre = new_staging_dir(project)
ev_path = os.path.join(staging_dir_pre, "evidence.jsonl")
ev = EvidenceLog(
ev_path,
max_chars=int(cfg.get("evidence_max_chars", 4000) or 4000),
redact=bool(cfg.get("redact_secrets", True)),
)
evidence.attach(backend, ev)
ev.log("cycle", "start", night=night, project=project,
backend=backend.name, model=cfg.get("model", ""),
config={k: cfg.get(k) for k in (
"backend", "model", "optimizer_backend", "optimizer_model",
"target_backend", "target_model", "gate_mode", "gate_metric",
"gate_mixed_weight", "edit_budget", "holdout_fraction",
"dream_rollouts", "dream_factor", "recall_k",
"max_tasks_per_night", "lookback_hours", "llm_mine",
"evolve_skill", "evolve_memory")})
# ── live skill/memory docs ───────────────────────────────────────────
live_memory_path = os.path.join(project, "CLAUDE.md")
live_skill_path = cfg.managed_skill_path()
@@ -175,6 +212,17 @@ def run_sleep_cycle(
)
n_sessions = len(digests)
_progress(cfg, f"harvest done: sessions={n_sessions}")
if ev is not None:
# The transcript end of the evidentiary chain: which sessions were
# even considered, and what signals they carried into mining.
for d in digests:
ev.log("harvest", "session", session_id=d.session_id,
project=d.project,
n_user_prompts=len(d.user_prompts),
user_prompts_head=[p[:200] for p in d.user_prompts[:6]],
assistant_final_head=(d.assistant_finals[-1][:300]
if d.assistant_finals else ""),
feedback_signals=list(d.feedback_signals or []))
# When a real backend is configured, use it to mine checkable tasks from
# the transcripts (rubric/rule judges); otherwise fall back to the
# heuristic miner (no API, no checkable reference).
@@ -206,6 +254,17 @@ def run_sleep_cycle(
)
_progress(cfg, f"mine done: tasks={len(tasks)}")
if ev is not None:
# Final task pool with split assignment: which tasks train the edits
# vs. which held-out tasks gate them (works for seeded tasks too).
for t in tasks:
ev.log("mine", "task_ready", task_id=t.id, split=t.split,
origin=t.origin, intent=t.intent[:300],
reference_kind=t.reference_kind,
checks=(t.judge or {}).get("checks", []),
rubric=(t.reference if t.reference_kind == "rubric" else ""),
source_sessions=list(t.source_sessions or []))
report = SleepReport(
night=night, project=project, started_at=started,
n_sessions=n_sessions, n_tasks=len(tasks),
@@ -218,6 +277,9 @@ def run_sleep_cycle(
state.record_night({"night": night, "accepted": False, "n_tasks": 0})
if not dry_run:
state.save()
if ev is not None:
ev.log("cycle", "end", night=night, outcome="no_tasks",
tokens_used=backend.tokens_used())
staging_dir = ""
return CycleOutcome(report, staging_dir, False, [])
@@ -281,7 +343,13 @@ def run_sleep_cycle(
live_skill_path=live_skill_path,
live_memory_path=live_memory_path,
report_md=report_md,
out_dir=staging_dir_pre,
)
if ev is not None:
ev.log("stage", "staged", staging_dir=staging_dir,
has_skill=proposed_skill is not None,
has_memory=proposed_memory is not None,
accepted=result.accepted)
# Observability: persist per-task held-out evidence + optimizer/codex errors so a
# 0.0->0.0 night self-explains (empty responses vs failing checks vs no edits) — the
# cycle previously captured none of this, making the gate a black box (#learning-stall).
@@ -321,4 +389,13 @@ def run_sleep_cycle(
adopted = bool(adopted_paths)
state.save()
if ev is not None:
ev.log("cycle", "end", night=night, outcome="completed",
gate_action=report.gate_action, accepted=report.accepted,
baseline_score=report.baseline_score,
candidate_score=report.candidate_score,
n_applied_edits=len(report.edits),
n_rejected_edits=len(report.rejected_edits),
tokens_used=report.tokens_used, adopted=adopted)
return CycleOutcome(report, staging_dir, adopted, adopted_paths)

133
skillopt_sleep/evidence.py Normal file
View File

@@ -0,0 +1,133 @@
"""SkillOpt-Sleep — the per-night evidentiary chain (``evidence.jsonl``).
The existing report/diagnostics answer *what* the cycle decided; they do not
answer *why*. This module records the full causal chain, per night:
transcript session -> miner exchange (prompt + raw reply)
-> mined task (+ its checks, + source session ids)
-> split assignment (train / val)
-> every replay attempt (phase-tagged, full prompt
and response, cache hits marked)
-> per-task scores with the failing checks named
-> the reflect exchange (prompt, raw reply, parsed
edits) and every gate trial
-> the final gate decision with the score arithmetic
-> what was staged
Design constraints (matching the sleep engine's contract):
* pure stdlib, thread-safe (replay batches run in a thread pool);
* every persisted string passes through ``redact_secrets`` and a length
cap, so the log can never leak more than diagnostics already could;
* append-only JSONL so a crashed night still leaves its partial chain;
* zero behavior change when disabled (``evidence_log: false``).
Events share the shape::
{"ts": <iso8601>, "seq": <int>, "stage": <str>, "event": <str>, ...}
"""
from __future__ import annotations
import json
import os
import threading
import time
from typing import Any, Optional
from skillopt_sleep.staging import redact_secrets
def _now_iso() -> str:
return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime())
class EvidenceLog:
"""Append-only, thread-safe JSONL logger for one sleep night."""
def __init__(self, path: str, *, max_chars: int = 4000, redact: bool = True) -> None:
self.path = path
self.max_chars = max(200, int(max_chars))
self.redact = redact
self._lock = threading.Lock()
self._seq = 0
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
# ── sanitization ──────────────────────────────────────────────────────
def _clean(self, value: Any) -> Any:
if isinstance(value, str):
if len(value) > self.max_chars:
dropped = len(value) - self.max_chars
value = value[: self.max_chars] + f"…[truncated {dropped} chars]"
return redact_secrets(value) if self.redact else value
if isinstance(value, dict):
return {k: self._clean(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [self._clean(v) for v in value]
return value
# ── the one write path ────────────────────────────────────────────────
def log(self, stage: str, event: str, **data: Any) -> None:
record = {"ts": _now_iso(), "stage": stage, "event": event}
record.update(self._clean(data))
with self._lock:
self._seq += 1
record["seq"] = self._seq
try:
line = json.dumps(record, ensure_ascii=False, default=str)
with open(self.path, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception:
# Evidence must never break a night; drop the record instead.
pass
def attach(backend, ev: Optional[EvidenceLog]) -> None:
"""Attach ``ev`` to a backend — and, for DualBackend, to both halves —
so every layer that wants to log can find it via ``backend.evidence``."""
if backend is None:
return
backend.evidence = ev
for half in ("target", "optimizer"):
sub = getattr(backend, half, None)
if sub is not None:
sub.evidence = ev
def get(backend) -> Optional[EvidenceLog]:
return getattr(backend, "evidence", None)
def set_phase(backend, phase: str) -> None:
"""Tag subsequent replay calls with a phase label (baseline_val,
train, gate_trial:skill, final_val, ...). Phases are sequential in the
consolidation loop, so a plain attribute is safe; parallelism only ever
happens *within* one phase."""
if backend is None:
return
backend.evidence_phase = phase
for half in ("target", "optimizer"):
sub = getattr(backend, half, None)
if sub is not None:
sub.evidence_phase = phase
def phase(backend) -> str:
return getattr(backend, "evidence_phase", "") or ""
def read_events(path: str) -> list:
"""Best-effort reader for the dashboard: skips corrupt lines."""
out = []
try:
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
out.append(json.loads(line))
except Exception:
continue
except OSError:
return []
out.sort(key=lambda r: r.get("seq", 0))
return out

81
skillopt_sleep/llm_miner.py
View File

@@ -22,51 +22,22 @@ import json
import re
from typing import Any, Callable, Dict, List
from skillopt_sleep import prompts as prompt_registry
from skillopt_sleep.backend import Backend, _extract_json
from skillopt_sleep.types import SessionDigest, TaskRecord
_MINER_PROMPT = """You are mining a user's past AI-assistant sessions to find RECURRING tasks
worth optimizing a skill for. From the session below, extract 0-3 reusable tasks.
A good task is something the user asks for repeatedly or had to correct, where a
GENERAL rule would help next time (formatting, structure, tool-use, conventions).
Skip one-off or purely exploratory requests.
For each task return:
- "intent": the reusable request, generalized (no one-off specifics)
- "checks": a list of programmatic success checks a grader can run on a future
answer. Each check is one of:
{"op":"section_present","arg":"<heading text>"}
{"op":"regex","arg":"<python regex the answer must match>"}
{"op":"contains","arg":"<substring the answer must contain>"}
{"op":"max_chars","arg":<int>}
Only include checks you are confident a GOOD answer must satisfy.
- "rubric": a one-sentence description of what a good answer looks like
- "satisfied": true/false — did the user seem satisfied with the assistant's answer?
Return ONLY a JSON array (possibly empty). No prose.
# Session
project: __PROJECT__
user prompts:
__PROMPTS__
assistant final (last):
__FINAL__
feedback signals: __FEEDBACK__
"""
def _digest_to_prompt(d: SessionDigest) -> str:
# Template lives in the central prompt registry (skillopt_sleep.prompts)
# so the dashboard can display and override it live.
prompts = "\n".join(f" - {p[:240]}" for p in d.user_prompts[:6]) or " (none)"
final = (d.assistant_finals[-1][:400] if d.assistant_finals else "(none)")
return (
_MINER_PROMPT
.replace("__PROJECT__", d.project or "(unknown)")
.replace("__PROMPTS__", prompts)
.replace("__FINAL__", final)
.replace("__FEEDBACK__", ", ".join(d.feedback_signals[:6]) or "(none)")
)
return prompt_registry.render("miner", {
"__PROJECT__": d.project or "(unknown)",
"__PROMPTS__": prompts,
"__FINAL__": final,
"__FEEDBACK__": ", ".join(d.feedback_signals[:6]) or "(none)",
})
def _mk_task(d: SessionDigest, obj: Dict[str, Any], idx: int) -> TaskRecord | None:
@@ -114,21 +85,45 @@ def make_llm_miner(
"""Return an llm_miner(digests) -> list[TaskRecord] bound to a backend."""
def _miner(digests: List[SessionDigest]) -> List[TaskRecord]:
ev = getattr(backend, "evidence", None)
out: List[TaskRecord] = []
for d in digests[:max_sessions]:
if not d.user_prompts:
continue
raw = backend._call(_digest_to_prompt(d), max_tokens=800) # type: ignore[attr-defined]
prompt = _digest_to_prompt(d)
raw = backend._call(prompt, max_tokens=800) # type: ignore[attr-defined]
arr = _extract_json(raw, "array")
if not isinstance(arr, list):
continue
for i, obj in enumerate(arr[:3]):
candidates = arr if isinstance(arr, list) else []
made: List[TaskRecord] = []
dropped = 0
full = False
for i, obj in enumerate(candidates[:3]):
if isinstance(obj, dict):
t = _mk_task(d, obj, i)
if t is not None:
made.append(t)
out.append(t)
else:
dropped += 1 # not checkable -> dropped, and now logged
if len(out) >= max_tasks:
return out
full = True
break
if ev is not None:
# The transcript->task link of the evidentiary chain: what this
# session was, exactly what the miner was asked, exactly what it
# replied, and which TaskRecords (with checks) came out of it.
ev.log("mine", "miner_exchange", session_id=d.session_id,
project=d.project, prompt=prompt, raw_reply=raw,
parse_ok=isinstance(arr, list), n_candidates=len(candidates),
n_tasks=len(made), n_dropped_uncheckable=dropped)
for t in made:
ev.log("mine", "task_mined", task_id=t.id,
session_id=d.session_id, intent=t.intent,
reference_kind=t.reference_kind,
checks=(t.judge or {}).get("checks", []),
rubric=t.reference, outcome=t.outcome)
if full:
return out
return out
return _miner

237
skillopt_sleep/prompts.py Normal file
View File

@@ -0,0 +1,237 @@
"""SkillOpt-Sleep — central prompt registry with live user overrides.
Every LLM-facing prompt template the sleep cycle uses (miner / attempt /
judge / reflect) lives here, in one place, instead of being scattered as
inline literals. Two consequences:
1. **Auditability** — the dashboard (and any human) can display exactly
what instructions each agent role receives, per stage.
2. **Live tuning** — a user override file (``prompts.json`` in the state
dir, or ``SKILLOPT_SLEEP_PROMPTS_PATH``) replaces any template without
touching code. The file's mtime is checked on every read, so an edit
made while a cycle is running takes effect on the very next call.
Placeholders use the ``__NAME__`` convention (simple ``str.replace``, no
``str.format``) because the templates themselves contain JSON braces.
The default texts are byte-for-byte the prompts previously inlined in
``backend.py`` / ``llm_miner.py``, so behavior is unchanged unless the user
overrides a template.
"""
from __future__ import annotations
import json
import os
import threading
from typing import Dict, List, Optional
from skillopt_sleep.config import HOME_STATE_DIR
# ── default templates ─────────────────────────────────────────────────────────
_MINER = """You are mining a user's past AI-assistant sessions to find RECURRING tasks
worth optimizing a skill for. From the session below, extract 0-3 reusable tasks.
A good task is something the user asks for repeatedly or had to correct, where a
GENERAL rule would help next time (formatting, structure, tool-use, conventions).
Skip one-off or purely exploratory requests.
For each task return:
- "intent": the reusable request, generalized (no one-off specifics)
- "checks": a list of programmatic success checks a grader can run on a future
answer. Each check is one of:
{"op":"section_present","arg":"<heading text>"}
{"op":"regex","arg":"<python regex the answer must match>"}
{"op":"contains","arg":"<substring the answer must contain>"}
{"op":"max_chars","arg":<int>}
Only include checks you are confident a GOOD answer must satisfy.
- "rubric": a one-sentence description of what a good answer looks like
- "satisfied": true/false — did the user seem satisfied with the assistant's answer?
Return ONLY a JSON array (possibly empty). No prose.
# Session
project: __PROJECT__
user prompts:
__PROMPTS__
assistant final (last):
__FINAL__
feedback signals: __FEEDBACK__
"""
_ATTEMPT = (
"Complete the following task for the user. Follow the skill and memory "
"guidance below, including any output-format and length requirements. "
"When a 'Learned preferences' rule sets an explicit limit (e.g. a length "
"cap), prefer that rule over more general advice it refines.\n\n"
"# Skill\n__SKILL__\n\n# Memory\n__MEMORY__\n\n"
"# Task\n__INTENT__\n\n__CONTEXT__\n\n"
"Return ONLY the final answer text, nothing else."
)
_JUDGE = (
"Score how well the response satisfies the rubric, 0..1. "
'Return ONLY JSON {"score": <0..1>, "reason": "..."}.\n\n'
"# Rubric\n__RUBRIC__\n\n# Response\n__RESPONSE__"
)
_REFLECT = (
"You are SkillOpt's optimizer. The agent keeps failing the recurring "
"tasks below. Propose at most __EDIT_BUDGET__ bounded edits to the "
"__TARGET__ document so it stops failing. Each edit MUST be a short, "
"GENERAL, reusable rule or preference (never task-specific, never an "
"answer to a single task). If exact failing criteria are listed, your "
"edits MUST make future outputs satisfy every one of them.\n"
"BE CONCRETE: quote the exact threshold, section name, or format from "
"the criteria verbatim in your rule (e.g. write 'keep the entire "
"response under 1200 characters', NOT 'respect length limits'). Vague "
"rules do not change behavior; specific numeric/structural rules do.\n"
"IMPORTANT: your edits are APPENDED to a 'Learned preferences' block; "
"you CANNOT delete the existing instructions above. If the current "
"__TARGET__ text conflicts with a criterion (e.g. it says 'be exhaustive' "
"but outputs must be under a character limit), write an explicit, "
"forceful OVERRIDE rule stating it supersedes the conflicting "
"instruction, and put the hard requirement first.\n"
"HARD CONSTRAINT: every rule you write MUST be consistent with the "
"'Task output contract' below (if shown). NEVER propose a rule that "
"changes the required output format/language, tells the agent to ask "
"the user a question, or otherwise violates that contract — such a "
"rule scores ZERO because the evaluator cannot honor it.\n"
'Return ONLY a JSON array: '
'[{"op":"add|replace|delete","content":"<rule>","anchor":"<text to replace/delete, optional>","rationale":"<why>"}].\n\n'
"# Current __TARGET__\n__CUR_DOC__\n"
"__GUARD__"
"__CRITERIA__\n"
"__PREFS__\n\n"
"# Recurring failures\n__FAILURES__"
)
# name -> {text, stage, role, description, placeholders}
DEFAULTS: Dict[str, Dict] = {
"miner": {
"text": _MINER,
"stage": "mine",
"role": "optimizer",
"description": "Turns one harvested session digest into 0-3 checkable TaskRecords.",
"placeholders": ["__PROJECT__", "__PROMPTS__", "__FINAL__", "__FEEDBACK__"],
},
"attempt": {
"text": _ATTEMPT,
"stage": "replay",
"role": "target",
"description": "The clean-context rollout: solve a mined task given only skill+memory.",
"placeholders": ["__SKILL__", "__MEMORY__", "__INTENT__", "__CONTEXT__"],
},
"judge": {
"text": _JUDGE,
"stage": "replay",
"role": "optimizer",
"description": "Rubric grading for tasks with no programmatic checks (0..1 JSON score).",
"placeholders": ["__RUBRIC__", "__RESPONSE__"],
},
"reflect": {
"text": _REFLECT,
"stage": "reflect",
"role": "optimizer",
"description": "Proposes bounded skill/memory edits from the recurring failures.",
"placeholders": [
"__EDIT_BUDGET__", "__TARGET__", "__CUR_DOC__", "__GUARD__",
"__CRITERIA__", "__PREFS__", "__FAILURES__",
],
},
}
# ── override file (mtime-cached; edits take effect on the next call) ──────────
_lock = threading.Lock()
_cache: Dict[str, object] = {"path": None, "mtime": None, "data": {}}
def overrides_path() -> str:
return os.environ.get("SKILLOPT_SLEEP_PROMPTS_PATH", "") or os.path.join(
HOME_STATE_DIR, "prompts.json"
)
def load_overrides() -> Dict[str, str]:
"""Return {name: replacement_text}, re-reading the file iff it changed."""
path = overrides_path()
with _lock:
try:
mtime = os.path.getmtime(path)
except OSError:
_cache.update(path=path, mtime=None, data={})
return {}
if _cache["path"] == path and _cache["mtime"] == mtime:
return dict(_cache["data"]) # type: ignore[arg-type]
try:
with open(path, encoding="utf-8") as f:
raw = json.load(f)
data = {
k: v for k, v in raw.items()
if k in DEFAULTS and isinstance(v, str) and v.strip()
}
except Exception:
data = {}
_cache.update(path=path, mtime=mtime, data=data)
return dict(data)
def save_overrides(overrides: Dict[str, Optional[str]]) -> Dict[str, str]:
"""Merge ``overrides`` into the override file. A None/empty value removes
that override (reverting the template to its default). Returns the new
effective override map."""
path = overrides_path()
current = load_overrides()
for k, v in overrides.items():
if k not in DEFAULTS:
continue
if v is None or not str(v).strip():
current.pop(k, None)
else:
current[k] = str(v)
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(current, f, ensure_ascii=False, indent=2)
with _lock:
_cache.update(path=None, mtime=None, data={}) # force re-read
return current
def get_prompt(name: str) -> str:
"""Effective template text for ``name`` (override if present, else default)."""
ov = load_overrides()
if name in ov:
return ov[name]
return DEFAULTS[name]["text"]
def is_overridden(name: str) -> bool:
return name in load_overrides()
def render(name: str, mapping: Dict[str, str]) -> str:
"""Substitute ``__NAME__`` placeholders via str.replace (format-safe)."""
text = get_prompt(name)
for k, v in mapping.items():
text = text.replace(k, v)
return text
def describe() -> List[Dict]:
"""Registry snapshot for the dashboard: defaults + active overrides."""
ov = load_overrides()
out = []
for name, meta in DEFAULTS.items():
out.append({
"name": name,
"stage": meta["stage"],
"role": meta["role"],
"description": meta["description"],
"placeholders": meta["placeholders"],
"default": meta["text"],
"override": ov.get(name),
"effective": ov.get(name) or meta["text"],
})
return out

12
skillopt_sleep/replay.py
View File

@@ -54,6 +54,18 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
else:
hard, soft, rationale = backend.judge(task, response)
ev = getattr(backend, "evidence", None)
if ev is not None:
# One scored-attempt record per (phase, task): the task->score link of
# the evidentiary chain, with the failing checks named in `why`.
ev.log("replay", "result",
phase=getattr(backend, "evidence_phase", ""),
task_id=task.id, split=task.split, origin=task.origin,
reference_kind=task.reference_kind, sample_id=sample_id,
hard=float(hard), soft=float(soft), why=rationale or "",
response_head=(response or "")[:400], tools_called=tools_called,
tokens=int(tokens), latency_ms=round(latency_ms, 1))
return ReplayResult(
id=task.id,
hard=float(hard),

27
skillopt_sleep/staging.py
View File

@@ -80,6 +80,16 @@ def staging_root(project: str) -> str:
return os.path.join(project, ".skillopt-sleep", "staging")
def new_staging_dir(project: str) -> str:
"""A staging path that is unique even for two runs in the same second."""
base = os.path.join(staging_root(project), _ts_dir())
out, i = base, 2
while os.path.exists(out):
out = f"{base}-{i}"
i += 1
return out
def latest_staging(project: str) -> Optional[str]:
root = staging_root(project)
if not os.path.isdir(root):
@@ -89,7 +99,12 @@ def latest_staging(project: str) -> Optional[str]:
key=lambda p: os.path.getmtime(p),
reverse=True,
)
return subs[0] if subs else None
for p in subs:
# Only adoptable folders count: a no-tasks night leaves evidence.jsonl
# but no manifest, and adopt() needs the manifest.
if os.path.exists(os.path.join(p, "manifest.json")):
return p
return None
def write_staging(
@@ -101,9 +116,15 @@ def write_staging(
live_skill_path: str,
live_memory_path: str,
report_md: str,
out_dir: str = "",
) -> str:
"""Write proposals + report into staging/<ts>/ and return that path."""
out = os.path.join(staging_root(project), _ts_dir())
"""Write proposals + report into staging/<ts>/ and return that path.
``out_dir`` lets the cycle pre-create the night's staging folder at cycle
START, so incremental artifacts (evidence.jsonl) accumulate in the same
place the report lands.
"""
out = out_dir or os.path.join(staging_root(project), _ts_dir())
os.makedirs(out, exist_ok=True)
manifest = {

157
tests/test_sleep_evidence.py Normal file
View File

@@ -0,0 +1,157 @@
"""Tests for the evidence log and the prompt registry.
Pure-stdlib (unittest), deterministic, no API key, no network,
no third-party deps.
Run: python -m unittest tests.test_sleep_evidence
"""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from unittest import mock
from skillopt_sleep import prompts as prompt_registry
from skillopt_sleep.config import load_config
from skillopt_sleep.cycle import run_sleep_cycle
from skillopt_sleep.evidence import EvidenceLog, read_events
from skillopt_sleep.experiments.personas import researcher_persona
from skillopt_sleep.mine import assign_splits
def _events_by(events, stage=None, event=None):
out = events
if stage is not None:
out = [e for e in out if e.get("stage") == stage]
if event is not None:
out = [e for e in out if e.get("event") == event]
return out
class TestEvidenceLog(unittest.TestCase):
def test_append_redact_truncate_and_order(self):
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "e.jsonl")
ev = EvidenceLog(path, max_chars=200)
ev.log("replay", "model_call", prompt="x" * 500,
secret="api_key=sk-abcdefghijklmnop")
ev.log("gate", "decision", action="accept")
events = read_events(path)
self.assertEqual([e["seq"] for e in events], [1, 2])
self.assertIn("truncated", events[0]["prompt"])
self.assertLessEqual(len(events[0]["prompt"]), 260)
self.assertNotIn("sk-abcdefghijklmnop", json.dumps(events))
self.assertIn("REDACTED", events[0]["secret"])
def test_reader_skips_corrupt_lines(self):
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "e.jsonl")
ev = EvidenceLog(path)
ev.log("cycle", "start")
with open(path, "a", encoding="utf-8") as f:
f.write("{not json\n")
ev.log("cycle", "end")
self.assertEqual(len(read_events(path)), 2)
class TestPromptRegistry(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.patch = mock.patch.dict(os.environ, {
"SKILLOPT_SLEEP_PROMPTS_PATH": os.path.join(self.tmp.name, "prompts.json")})
self.patch.start()
def tearDown(self):
self.patch.stop()
self.tmp.cleanup()
def test_defaults_match_legacy_prompts(self):
# The registry must reproduce the exact legacy wording by default.
self.assertIn("You are SkillOpt's optimizer", prompt_registry.get_prompt("reflect"))
self.assertIn("RECURRING tasks", prompt_registry.get_prompt("miner"))
self.assertIn("Return ONLY the final answer text", prompt_registry.get_prompt("attempt"))
rendered = prompt_registry.render("attempt", {
"__SKILL__": "S", "__MEMORY__": "M", "__INTENT__": "I", "__CONTEXT__": "C"})
self.assertIn("# Skill\nS", rendered)
self.assertNotIn("__SKILL__", rendered)
def test_override_takes_effect_without_restart(self):
self.assertFalse(prompt_registry.is_overridden("judge"))
prompt_registry.save_overrides({"judge": "CUSTOM __RUBRIC__ / __RESPONSE__"})
self.assertTrue(prompt_registry.is_overridden("judge"))
self.assertEqual(
prompt_registry.render("judge", {"__RUBRIC__": "r", "__RESPONSE__": "x"}),
"CUSTOM r / x")
# empty value reverts to default
prompt_registry.save_overrides({"judge": None})
self.assertFalse(prompt_registry.is_overridden("judge"))
self.assertIn("Score how well", prompt_registry.get_prompt("judge"))
def test_unknown_names_are_ignored(self):
out = prompt_registry.save_overrides({"nope": "x", "miner": "M __PROMPTS__"})
self.assertEqual(set(out), {"miner"})
prompt_registry.save_overrides({"miner": ""})
class TestCycleEvidence(unittest.TestCase):
def _run(self, **cfg_extra):
proj = tempfile.mkdtemp()
home = tempfile.mkdtemp()
cfg = load_config(
invoked_project=proj, projects="invoked", backend="mock",
claude_home=os.path.join(home, ".claude"), auto_adopt=False,
**cfg_extra)
tasks = assign_splits(researcher_persona(), holdout_fraction=0.34, seed=42)
outcome = run_sleep_cycle(cfg, seed_tasks=tasks)
return outcome
def test_evidence_written_with_full_chain(self):
outcome = self._run()
path = os.path.join(outcome.staging_dir, "evidence.jsonl")
self.assertTrue(os.path.exists(path), "evidence.jsonl missing from staging dir")
events = read_events(path)
# chain: cycle start .. task_ready .. replay results (phased) ..
# reflect edits .. gate baseline/trial/decision .. staged .. cycle end
self.assertTrue(_events_by(events, "cycle", "start"))
self.assertTrue(_events_by(events, "mine", "task_ready"))
splits = {e["split"] for e in _events_by(events, "mine", "task_ready")}
self.assertIn("train", splits)
results = _events_by(events, "replay", "result")
self.assertTrue(results)
phases = {e["phase"] for e in results}
self.assertIn("baseline_val", phases)
self.assertIn("final_val", phases)
self.assertTrue(_events_by(events, "reflect", "edits_returned"))
self.assertTrue(_events_by(events, "gate", "baseline"))
decision = _events_by(events, "gate", "decision")
self.assertEqual(len(decision), 1)
self.assertIn("formula", decision[0])
self.assertTrue(_events_by(events, "stage", "staged"))
end = _events_by(events, "cycle", "end")
self.assertEqual(len(end), 1)
self.assertEqual(end[0]["outcome"], "completed")
# the report landed in the SAME pre-created folder as the evidence
self.assertTrue(os.path.exists(os.path.join(outcome.staging_dir, "report.md")))
def test_evidence_can_be_disabled(self):
outcome = self._run(evidence_log=False)
self.assertFalse(
os.path.exists(os.path.join(outcome.staging_dir, "evidence.jsonl")))
def test_no_tasks_night_is_not_adoptable_but_keeps_evidence(self):
from skillopt_sleep.staging import latest_staging
proj = tempfile.mkdtemp()
home = tempfile.mkdtemp()
cfg = load_config(
invoked_project=proj, projects="invoked", backend="mock",
claude_home=os.path.join(home, ".claude"))
outcome = run_sleep_cycle(cfg, seed_tasks=[])
self.assertEqual(outcome.staging_dir, "")
# an evidence-only folder exists but latest_staging must skip it
self.assertIsNone(latest_staging(proj))
if __name__ == "__main__":
unittest.main()