diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index 5a761b66..a2fa05a3 100644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -76,7 +76,8 @@ def _add_common(p: argparse.ArgumentParser) -> None: p.add_argument("--codex-path", default="", help="path to the real @openai/codex binary") p.add_argument("--claude-home", default="", help="override ~/.claude (also isolates state)") p.add_argument("--codex-home", default="", help="override ~/.codex for archived session harvest") - p.add_argument("--source", default="", choices=["", "claude", "codex", "auto"], + p.add_argument("--source", default="", + choices=["", "claude", "codex", "antigravity", "auto"], help="session transcript source") p.add_argument("--lookback-hours", type=int, default=None, help="harvest window in hours; 0 = scan full history") @@ -513,6 +514,12 @@ def main(argv=None) -> int: p_unsched = sub.add_parser("unschedule", help="remove the nightly cron entry") _add_common(p_unsched) p_unsched.add_argument("--all", action="store_true", help="remove all managed entries") + p_dash = sub.add_parser( + "dashboard", help="serve the local control-panel dashboard (127.0.0.1)") + _add_common(p_dash) + p_dash.add_argument("--port", type=int, default=8321) + p_dash.add_argument("--no-browser", action="store_true", + help="don't auto-open the browser") args = parser.parse_args(argv) if args.cmd == "run": @@ -529,6 +536,10 @@ def main(argv=None) -> int: return cmd_schedule(args) if args.cmd == "unschedule": return cmd_unschedule(args) + if args.cmd == "dashboard": + from skillopt_sleep.dashboard import serve + return serve(project=args.project or os.getcwd(), port=args.port, + open_browser=not args.no_browser) parser.print_help() return 2 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..63eb66a2 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -25,7 +25,8 @@ # ── scope ────────────────────────────────────────────────────────────── "claude_home": CLAUDE_HOME, "codex_home": CODEX_HOME, - "transcript_source": "claude", # "claude" | "codex" | "auto" + "transcript_source": "claude", # "claude" | "codex" | "antigravity" | "auto" + "antigravity_conversations_dir": "", # "" => ~/.gemini/antigravity/conversations "projects": "invoked", # "invoked" | "all" | [list of abs paths] "invoked_project": "", # filled at runtime (cwd) when projects == "invoked" "lookback_hours": 72, # harvest window when no prior sleep recorded @@ -38,6 +39,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 +64,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..524760b5 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,21 @@ 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: the COMPLETE + # SessionDigest for every considered session (per-field truncation + # is handled by the evidence log's max_chars cap). + for d in digests: + ev.log("harvest", "session", session_id=d.session_id, + project=d.project, source=cfg.get("transcript_source"), + started_at=d.started_at, ended_at=d.ended_at, + raw_path=d.raw_path, + n_user_prompts=len(d.user_prompts), + user_prompts=list(d.user_prompts), + assistant_finals=list(d.assistant_finals), + tools_used=list(d.tools_used or []), + files_touched=list(d.files_touched or []), + 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 +257,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 +280,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 +346,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 +392,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/dashboard.html b/skillopt_sleep/dashboard.html new file mode 100644 index 00000000..6b4b6e93 --- /dev/null +++ b/skillopt_sleep/dashboard.html @@ -0,0 +1,598 @@ + + + + + +SkillOpt-Sleep — Control Panel + + + +
+ +
+
idle
+ + +
+ +
+ + +
+ +

Pipeline — one sleep night, left to right

+
+ + +
+
⚙Models & Knobs + + + saved ✓
+
+
+
TARGET the model whose skill is being evolved — runs every attempt (replay rollout)
+
OPTIMIZER the model that mines tasks, judges rubrics and writes edits (reflect)
+
Leave the optimizer/target overrides empty to have the base backend play all roles. Changes apply from the next run. Prompt edits (below, per stage) apply to the very next model call.
+
+
+
+
+ +
+ + +
+
▤Run log
+ 
(no run this session)
+
+
+
+ + + + diff --git a/skillopt_sleep/dashboard.py b/skillopt_sleep/dashboard.py new file mode 100644 index 00000000..9df3ea06 --- /dev/null +++ b/skillopt_sleep/dashboard.py @@ -0,0 +1,296 @@ +"""SkillOpt-Sleep — local control-panel dashboard. + +A zero-dependency (stdlib ``http.server``) web UI over one project's sleep +pipeline. It is arranged to mirror the actual data flow — + + transcripts -> harvest -> mine -> split -> replay -> reflect -> gate + -> stage -> adopt + +— and for every stage shows: which agent role runs it (target / optimizer / +pure code), which model that role resolves to, the exact prompt template it +receives (editable, live), and the selected night's evidence events for that +stage (from ``evidence.jsonl``). Config changes are written to the user +config file and apply from the next run; prompt overrides apply to the very +next model call (the registry re-reads its override file on mtime change). + +Serves on 127.0.0.1 only. + + python -m skillopt_sleep dashboard [--project DIR] [--port N] +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Dict, List, Optional + +from skillopt_sleep import prompts as prompt_registry +from skillopt_sleep.config import DEFAULTS, HOME_STATE_DIR, load_config +from skillopt_sleep.evidence import read_events +from skillopt_sleep.staging import adopt as adopt_staging +from skillopt_sleep.staging import staging_root + +_HTML_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dashboard.html") + +# Config keys the dashboard may write (safety allowlist: everything else in +# the user config file is preserved untouched). +_EDITABLE_KEYS = { + "backend", "model", + "optimizer_backend", "optimizer_model", "target_backend", "target_model", + "azure_endpoint", + "gate_mode", "gate_metric", "gate_mixed_weight", + "edit_budget", "holdout_fraction", "lookback_hours", + "max_tasks_per_night", "max_sessions_per_night", "max_tokens_per_night", + "dream_rollouts", "dream_factor", "recall_k", + "evolve_skill", "evolve_memory", "llm_mine", "target_skill_path", + "preferences", "evidence_log", "evidence_max_chars", "auto_adopt", + "transcript_source", +} + + +def _read_json(path: str) -> Optional[Any]: + try: + with open(path, encoding="utf-8") as f: + return json.load(f) + except Exception: + return None + + +def _read_text(path: str, limit: int = 200_000) -> str: + try: + with open(path, encoding="utf-8") as f: + return f.read(limit) + except Exception: + return "" + + +def _user_config_file() -> str: + return os.path.join(HOME_STATE_DIR, "config.json") + + +def _write_config(updates: Dict[str, Any]) -> Dict[str, Any]: + path = _user_config_file() + current = _read_json(path) or {} + for k, v in updates.items(): + if k not in _EDITABLE_KEYS: + continue + if v is None or v == "": + # empty resets the key to the built-in default + current.pop(k, None) + else: + current[k] = v + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + json.dump(current, f, ensure_ascii=False, indent=2) + return current + + +def _list_nights(project: str) -> List[Dict[str, Any]]: + root = staging_root(project) + out: List[Dict[str, Any]] = [] + if not os.path.isdir(root): + return out + for name in sorted(os.listdir(root), reverse=True): + d = os.path.join(root, name) + if not os.path.isdir(d): + continue + report = _read_json(os.path.join(d, "report.json")) or {} + entry = { + "ts": name, + "night": report.get("night"), + "accepted": report.get("accepted"), + "gate_action": report.get("gate_action", ""), + "baseline": report.get("baseline_score"), + "candidate": report.get("candidate_score"), + "n_tasks": report.get("n_tasks"), + "n_sessions": report.get("n_sessions"), + "tokens_used": report.get("tokens_used"), + "has_report": bool(report), + "has_evidence": os.path.exists(os.path.join(d, "evidence.jsonl")), + "has_manifest": os.path.exists(os.path.join(d, "manifest.json")), + "adopted": os.path.isdir(os.path.join(d, "backup")), + } + out.append(entry) + return out + + +class _RunState: + """At most one pipeline subprocess at a time, log tailed to a file.""" + + def __init__(self) -> None: + self.proc: Optional[subprocess.Popen] = None + self.log_path = "" + self.mode = "" + self.lock = threading.Lock() + + def running(self) -> bool: + return self.proc is not None and self.proc.poll() is None + + def start(self, project: str, dry_run: bool) -> Dict[str, Any]: + with self.lock: + if self.running(): + return {"ok": False, "error": "a run is already in progress"} + cfg = load_config(invoked_project=project) + self.log_path = os.path.join(cfg.state_dir, "dashboard-run.log") + os.makedirs(os.path.dirname(self.log_path), exist_ok=True) + self.mode = "dry-run" if dry_run else "run" + cmd = [sys.executable, "-m", "skillopt_sleep", self.mode, + "--project", project, "--progress"] + log = open(self.log_path, "w", encoding="utf-8") + no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0 + self.proc = subprocess.Popen( + cmd, stdout=log, stderr=subprocess.STDOUT, + creationflags=no_window, cwd=project or None, + ) + return {"ok": True, "mode": self.mode} + + def status(self) -> Dict[str, Any]: + tail = "" + if self.log_path: + text = _read_text(self.log_path) + tail = text[-6000:] + rc = None + if self.proc is not None: + rc = self.proc.poll() + return {"running": self.running(), "returncode": rc, + "mode": self.mode, "tail": tail} + + +class DashboardHandler(BaseHTTPRequestHandler): + project: str = "" + run_state: _RunState + + # ── plumbing ────────────────────────────────────────────────────────── + def log_message(self, fmt: str, *args: Any) -> None: # quiet server + pass + + def _send(self, code: int, body: bytes, ctype: str) -> None: + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def _json(self, obj: Any, code: int = 200) -> None: + self._send(code, json.dumps(obj, ensure_ascii=False, default=str).encode("utf-8"), + "application/json; charset=utf-8") + + def _body(self) -> Dict[str, Any]: + try: + n = int(self.headers.get("Content-Length", "0") or "0") + raw = self.rfile.read(n) if n else b"{}" + obj = json.loads(raw.decode("utf-8") or "{}") + return obj if isinstance(obj, dict) else {} + except Exception: + return {} + + # ── GET ─────────────────────────────────────────────────────────────── + def do_GET(self) -> None: # noqa: N802 (http.server API) + path = self.path.split("?", 1)[0] + if path in {"/", "/index.html"}: + html = _read_text(_HTML_PATH, limit=5_000_000) + self._send(200, html.encode("utf-8"), "text/html; charset=utf-8") + return + if path == "/api/overview": + cfg = load_config(invoked_project=self.project) + effective = {k: cfg.get(k) for k in sorted(_EDITABLE_KEYS)} + self._json({ + "project": self.project, + "config": effective, + "defaults": {k: DEFAULTS.get(k) for k in sorted(_EDITABLE_KEYS)}, + "config_path": _user_config_file(), + "prompts": prompt_registry.describe(), + "prompts_path": prompt_registry.overrides_path(), + "nights": _list_nights(self.project), + }) + return + if path.startswith("/api/night/"): + ts = os.path.basename(path[len("/api/night/"):]) + d = os.path.join(staging_root(self.project), ts) + if not os.path.isdir(d): + self._json({"error": "unknown night"}, 404) + return + self._json({ + "ts": ts, + "dir": d, + "report": _read_json(os.path.join(d, "report.json")), + "manifest": _read_json(os.path.join(d, "manifest.json")), + "diagnostics": _read_json(os.path.join(d, "diagnostics.json")), + "report_md": _read_text(os.path.join(d, "report.md")), + "proposed_skill": _read_text(os.path.join(d, "proposed_SKILL.md")), + "proposed_memory": _read_text(os.path.join(d, "proposed_CLAUDE.md")), + "evidence": read_events(os.path.join(d, "evidence.jsonl")), + "adopted": os.path.isdir(os.path.join(d, "backup")), + }) + return + if path == "/api/run/status": + self._json(self.run_state.status()) + return + self._json({"error": "not found"}, 404) + + # ── POST ────────────────────────────────────────────────────────────── + def do_POST(self) -> None: # noqa: N802 + path = self.path.split("?", 1)[0] + body = self._body() + if path == "/api/config": + updates = body.get("updates") or {} + if not isinstance(updates, dict): + self._json({"error": "updates must be an object"}, 400) + return + saved = _write_config(updates) + cfg = load_config(invoked_project=self.project) + self._json({"ok": True, "saved": saved, + "config": {k: cfg.get(k) for k in sorted(_EDITABLE_KEYS)}}) + return + if path == "/api/prompts": + updates = body.get("updates") or {} + if not isinstance(updates, dict): + self._json({"error": "updates must be an object"}, 400) + return + prompt_registry.save_overrides(updates) + self._json({"ok": True, "prompts": prompt_registry.describe()}) + return + if path == "/api/run": + self._json(self.run_state.start(self.project, bool(body.get("dry_run")))) + return + if path == "/api/adopt": + ts = os.path.basename(str(body.get("ts", ""))) + d = os.path.join(staging_root(self.project), ts) + if not os.path.isdir(d): + self._json({"error": "unknown night"}, 404) + return + try: + updated = adopt_staging(d) + except Exception as exc: # surface, don't crash the server + self._json({"ok": False, "error": str(exc)}, 500) + return + self._json({"ok": True, "updated": updated}) + return + self._json({"error": "not found"}, 404) + + +def serve(project: str = "", port: int = 8321, open_browser: bool = True) -> int: + project = os.path.abspath(project or os.getcwd()) + handler = DashboardHandler + handler.project = project + handler.run_state = _RunState() + httpd = ThreadingHTTPServer(("127.0.0.1", port), handler) + url = f"http://127.0.0.1:{httpd.server_address[1]}/" + print(f"[sleep] dashboard for {project}\n[sleep] serving {url} (Ctrl+C to stop)") + if open_browser: + try: + import webbrowser + threading.Timer(0.4, webbrowser.open, args=(url,)).start() + except Exception: + pass + try: + httpd.serve_forever() + except KeyboardInterrupt: + pass + finally: + httpd.server_close() + return 0 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/harvest_antigravity.py b/skillopt_sleep/harvest_antigravity.py new file mode 100644 index 00000000..ac5e41a7 --- /dev/null +++ b/skillopt_sleep/harvest_antigravity.py @@ -0,0 +1,236 @@ +"""SkillOpt-Sleep — harvest Google Antigravity conversation stores. + +Antigravity persists each conversation as a SQLite "trajectory" database in +``~/.gemini/antigravity/conversations/.db``. The ``steps`` table holds +protobuf-encoded step payloads; without the proprietary schema we extract the +human-readable content with a conservative protobuf walker that collects +UTF-8 string fields: + + * step_type 14 -> user messages (the typed prompt, e.g. "/goal ...") + * step_type 5 -> artifact/answer content the agent produced + * step_type 33 -> tool calls (JSON with toolSummary/toolAction) + +That is enough to build the same ``SessionDigest`` the Claude/Codex +harvesters produce: user prompts, assistant finals, tools used, feedback +signals. Databases may be locked by a live Antigravity process, so each file +is copied to a temp path before opening (read-only URI otherwise). + +Heuristic by design: if Antigravity's schema changes, the walker degrades to +returning fewer strings — never to crashing the night (a session that yields +no user prompts is simply skipped, same as an empty transcript). +""" +from __future__ import annotations + +import json +import os +import re +import shutil +import sqlite3 +import tempfile +import time +from typing import List, Optional + +from skillopt_sleep.harvest import _detect_feedback, _is_meta_prompt +from skillopt_sleep.types import SessionDigest + +DEFAULT_CONVERSATIONS_DIR = os.path.expanduser( + "~/.gemini/antigravity/conversations") + +_USER_STEP_TYPES = {14} +_ARTIFACT_STEP_TYPES = {5} +_TOOL_STEP_TYPES = {33} + +_UUID_RE = re.compile( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") + +# Antigravity injects a system wrapper around /goal tasks, and user steps also +# carry a permission-history of tool echoes like ``read_url(github.com)`` — +# neither is the user's own words. +_BOILERPLATE_MARKERS = ( + "marked this task with /goal", + "The system will force you to continue", +) +_TOOL_ECHO_RE = re.compile(r"^[\w$.\\/-]+\([^()]*\)$") + + +# ── generic protobuf string extraction ──────────────────────────────────────── + +def _read_varint(buf: bytes, i: int): + val = 0 + shift = 0 + n = len(buf) + while i < n: + b = buf[i] + i += 1 + val |= (b & 0x7F) << shift + shift += 7 + if not b & 0x80: + return val, i + if shift > 63: + break + return None, i + + +def _proto_strings(buf: bytes, depth: int = 0, out: Optional[List[str]] = None) -> List[str]: + """Collect plausible UTF-8 string fields from a protobuf blob (schema-less).""" + if out is None: + out = [] + if depth > 6 or len(out) > 400: + return out + i, n = 0, len(buf) + while i < n: + tag, i = _read_varint(buf, i) + if tag is None: + break + wire = tag & 7 + if wire == 0: + _v, i = _read_varint(buf, i) + if _v is None: + break + elif wire == 1: + i += 8 + elif wire == 5: + i += 4 + elif wire == 2: + ln, i = _read_varint(buf, i) + if ln is None or ln < 0 or i + ln > n: + break + chunk = buf[i:i + ln] + i += ln + text = None + try: + text = chunk.decode("utf-8") + except UnicodeDecodeError: + text = None + if text is not None and len(text) >= 16 and _looks_natural(text): + out.append(text) + else: + # possibly a nested message — recurse; a failed walk just + # contributes nothing + _proto_strings(chunk, depth + 1, out) + else: # unknown/deprecated wire types: bail out of this blob + break + return out + + +def _looks_natural(text: str) -> bool: + """Keep human/markdown text; drop ids, uuids, base64 runs, file URIs.""" + t = text.strip() + if not t or _UUID_RE.match(t): + return False + if t.startswith(("file:///", "http://", "https://")) and " " not in t: + return False + if " " not in t and len(t) > 40: # long spaceless token: id/base64 + return False + letters = sum(c.isalpha() or c.isspace() for c in t) + return letters / max(1, len(t)) > 0.55 + + +# ── per-database digestion ──────────────────────────────────────────────────── + +def _clean_user_prompt(text: str) -> str: + t = text.strip() + for prefix in ("/goal ", "/task ", "/ask "): + if t.lower().startswith(prefix): + t = t[len(prefix):] + return t.strip() + + +def _digest_db(path: str, project: str) -> Optional[SessionDigest]: + tmp = os.path.join(tempfile.gettempdir(), + f"skillopt_agy_{os.path.basename(path)}") + try: + shutil.copy2(path, tmp) + except OSError: + return None + try: + con = sqlite3.connect(f"file:{tmp}?mode=ro", uri=True) + rows = con.execute( + "SELECT idx, step_type, step_payload FROM steps ORDER BY idx" + ).fetchall() + con.close() + except Exception: + return None + finally: + try: + os.unlink(tmp) + except OSError: + pass + + prompts: List[str] = [] + finals: List[str] = [] + tools: List[str] = [] + for _idx, stype, payload in rows: + blob = payload if isinstance(payload, bytes) else str(payload or "").encode() + if not blob: + continue + if stype in _USER_STEP_TYPES: + strs = [ + s for s in _proto_strings(blob) + if not s.startswith("{") + and not any(m in s for m in _BOILERPLATE_MARKERS) + and not _TOOL_ECHO_RE.match(s.strip()) + ] + if strs: + p = _clean_user_prompt(max(strs, key=len)) + if p and not _is_meta_prompt(p): + prompts.append(p) + elif stype in _ARTIFACT_STEP_TYPES: + strs = _proto_strings(blob) + # prefer the artifact body over its ArtifactMetadata JSON envelope + body = [s for s in strs if not s.lstrip().startswith("{")] + if body or strs: + finals.append(max(body or strs, key=len)) + elif stype in _TOOL_STEP_TYPES: + for s in _proto_strings(blob): + if s.startswith("{"): + try: + obj = json.loads(s) + name = obj.get("toolSummary") or obj.get("toolAction") + if name and name not in tools: + tools.append(str(name)) + except Exception: + pass + + if not prompts: + return None + mtime = os.path.getmtime(path) + iso = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(mtime)) + feedback = _detect_feedback(" \n".join(prompts)) + return SessionDigest( + session_id=os.path.splitext(os.path.basename(path))[0], + project=project, + started_at=iso, ended_at=iso, + user_prompts=prompts, + assistant_finals=finals[-3:], + tools_used=tools[:12], + feedback_signals=feedback, + ) + + +def harvest_antigravity( + conversations_dir: str = "", + *, + invoked_project: str = "", + since_iso: Optional[str] = None, + limit: int = 0, +) -> List[SessionDigest]: + """Digest the most recent Antigravity conversations (newest first).""" + root = os.path.expanduser(conversations_dir or DEFAULT_CONVERSATIONS_DIR) + if not os.path.isdir(root): + return [] + dbs = [os.path.join(root, f) for f in os.listdir(root) if f.endswith(".db")] + dbs.sort(key=os.path.getmtime, reverse=True) + out: List[SessionDigest] = [] + for path in dbs: + if limit and len(out) >= limit: + break + if since_iso: + mtime_iso = time.strftime( + "%Y-%m-%dT%H:%M:%S", time.localtime(os.path.getmtime(path))) + if mtime_iso < since_iso: + continue + d = _digest_db(path, project=invoked_project or "antigravity") + if d is not None: + out.append(d) + return out diff --git a/skillopt_sleep/harvest_sources.py b/skillopt_sleep/harvest_sources.py index 501aa285..d58aa538 100644 --- a/skillopt_sleep/harvest_sources.py +++ b/skillopt_sleep/harvest_sources.py @@ -13,6 +13,14 @@ def harvest_for_config(cfg, *, since_iso: Optional[str] = None, limit: int = 0) scope = cfg.get("projects", "invoked") invoked_project = cfg.get("invoked_project", "") + if source == "antigravity": + from skillopt_sleep.harvest_antigravity import harvest_antigravity + return harvest_antigravity( + cfg.get("antigravity_conversations_dir", ""), + invoked_project=invoked_project, + since_iso=since_iso, + limit=limit, + ) if source == "codex": return harvest_codex( cfg.codex_archived_sessions_dir, 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..cdeb6e26 --- /dev/null +++ b/tests/test_sleep_evidence.py @@ -0,0 +1,222 @@ +"""Tests for the evidence log, the prompt registry, and the dashboard API. + +Pure-stdlib (unittest), deterministic, no API key, no network beyond +127.0.0.1, no third-party deps. + +Run: python -m unittest tests.test_sleep_evidence +""" +from __future__ import annotations + +import http.client +import json +import os +import tempfile +import threading +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)) + + +class TestDashboardApi(unittest.TestCase): + @classmethod + def setUpClass(cls): + from skillopt_sleep.dashboard import DashboardHandler, _RunState + from http.server import ThreadingHTTPServer + cls.tmp = tempfile.TemporaryDirectory() + cls.env = mock.patch.dict(os.environ, { + "SKILLOPT_SLEEP_PROMPTS_PATH": os.path.join(cls.tmp.name, "prompts.json")}) + cls.env.start() + DashboardHandler.project = cls.tmp.name + DashboardHandler.run_state = _RunState() + cls.httpd = ThreadingHTTPServer(("127.0.0.1", 0), DashboardHandler) + cls.port = cls.httpd.server_address[1] + cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.httpd.shutdown() + cls.httpd.server_close() + cls.env.stop() + cls.tmp.cleanup() + + def _get(self, path): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5) + conn.request("GET", path) + r = conn.getresponse() + body = json.loads(r.read().decode("utf-8")) if "json" in r.getheader("Content-Type", "") else r.read() + conn.close() + return r.status, body + + def _post(self, path, obj): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5) + payload = json.dumps(obj).encode("utf-8") + conn.request("POST", path, body=payload, + headers={"Content-Type": "application/json"}) + r = conn.getresponse() + body = json.loads(r.read().decode("utf-8")) + conn.close() + return r.status, body + + def test_overview_and_html(self): + status, body = self._get("/api/overview") + self.assertEqual(status, 200) + self.assertEqual(body["project"], self.tmp.name) + self.assertEqual({p["name"] for p in body["prompts"]}, + {"miner", "attempt", "judge", "reflect"}) + status, html = self._get("/") + self.assertEqual(status, 200) + self.assertIn(b"Control Panel", html) + + def test_prompt_roundtrip(self): + status, body = self._post("/api/prompts", {"updates": {"miner": "X __PROMPTS__"}}) + self.assertEqual(status, 200) + mined = [p for p in body["prompts"] if p["name"] == "miner"][0] + self.assertEqual(mined["override"], "X __PROMPTS__") + self._post("/api/prompts", {"updates": {"miner": None}}) + + def test_unknown_night_404(self): + status, _body = self._get("/api/night/nope") + self.assertEqual(status, 404) + + +if __name__ == "__main__": + unittest.main()