From 782955276a075ff6a233d33cbe923951117a4023 Mon Sep 17 00:00:00 2001 From: Alpha Date: Sat, 18 Jul 2026 21:51:46 -0600 Subject: [PATCH] feat(sleep): per-night evidence chain and live prompt registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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. --- skillopt_sleep/backend.py | 92 ++++++------- skillopt_sleep/config.py | 11 ++ skillopt_sleep/consolidate.py | 51 +++++++- skillopt_sleep/cycle.py | 83 +++++++++++- skillopt_sleep/evidence.py | 133 +++++++++++++++++++ skillopt_sleep/llm_miner.py | 81 ++++++------ skillopt_sleep/prompts.py | 237 ++++++++++++++++++++++++++++++++++ skillopt_sleep/replay.py | 12 ++ skillopt_sleep/staging.py | 27 +++- tests/test_sleep_evidence.py | 157 ++++++++++++++++++++++ 10 files changed, 789 insertions(+), 95 deletions(-) create mode 100644 skillopt_sleep/evidence.py create mode 100644 skillopt_sleep/prompts.py create mode 100644 tests/test_sleep_evidence.py diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py index cd130a48..8c46256d 100644 --- a/skillopt_sleep/backend.py +++ b/skillopt_sleep/backend.py @@ -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 @@ def _call(self, prompt: str, *, max_tokens: int = 1024) -> str: 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 @@ def attempt(self, task: TaskRecord, skill: str, memory: str, 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 @@ def judge(self, task: TaskRecord, response: str) -> Tuple[float, float, str]: 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 @@ def _explain(c: str) -> str: # 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":"","anchor":"","rationale":""}].\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 @@ def _explain(c: str) -> str: ) 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 diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py index 4a1c992f..19e32c5c 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -38,6 +38,14 @@ # ── optimizer ────────────────────────────────────────────────────────── "backend": "mock", # "mock" | "claude" | "codex" | "copilot" "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 "edit_budget": 4, # textual learning rate (max edits/night) @@ -55,6 +63,9 @@ "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", diff --git a/skillopt_sleep/consolidate.py b/skillopt_sleep/consolidate.py index a9ea6625..60615bea 100644 --- a/skillopt_sleep/consolidate.py +++ b/skillopt_sleep/consolidate.py @@ -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 _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: # 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 _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: 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 _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: 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 _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: 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, diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index f199211d..82f8fc32 100644 --- a/skillopt_sleep/cycle.py +++ b/skillopt_sleep/cycle.py @@ -14,7 +14,9 @@ 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,15 +116,50 @@ 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", ""), + 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() @@ -174,6 +211,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). @@ -205,6 +253,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), @@ -217,6 +276,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, []) @@ -280,7 +342,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). @@ -320,4 +388,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) diff --git a/skillopt_sleep/evidence.py b/skillopt_sleep/evidence.py new file mode 100644 index 00000000..300227a1 --- /dev/null +++ b/skillopt_sleep/evidence.py @@ -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": , "seq": , "stage": , "event": , ...} +""" +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 diff --git a/skillopt_sleep/llm_miner.py b/skillopt_sleep/llm_miner.py index dd78c636..ddf07803 100644 --- a/skillopt_sleep/llm_miner.py +++ b/skillopt_sleep/llm_miner.py @@ -22,51 +22,22 @@ 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":""} - {"op":"regex","arg":""} - {"op":"contains","arg":""} - {"op":"max_chars","arg":} - 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 diff --git a/skillopt_sleep/prompts.py b/skillopt_sleep/prompts.py new file mode 100644 index 00000000..3553328e --- /dev/null +++ b/skillopt_sleep/prompts.py @@ -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":""} + {"op":"regex","arg":""} + {"op":"contains","arg":""} + {"op":"max_chars","arg":} + 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":"","anchor":"","rationale":""}].\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 diff --git a/skillopt_sleep/replay.py b/skillopt_sleep/replay.py index e15f3dfe..730e6147 100644 --- a/skillopt_sleep/replay.py +++ b/skillopt_sleep/replay.py @@ -53,6 +53,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), diff --git a/skillopt_sleep/staging.py b/skillopt_sleep/staging.py index 49dd859b..3b2434d4 100644 --- a/skillopt_sleep/staging.py +++ b/skillopt_sleep/staging.py @@ -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// and return that path.""" - out = os.path.join(staging_root(project), _ts_dir()) + """Write proposals + report into staging// 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 = { diff --git a/tests/test_sleep_evidence.py b/tests/test_sleep_evidence.py new file mode 100644 index 00000000..ff5a6cf3 --- /dev/null +++ b/tests/test_sleep_evidence.py @@ -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()