Merge pull request #151 from Alphaxalchemy/feat/sleep-evidence-chain

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

Author: Yif-Yang

CHANGELOG.md

@@ -7,6 +7,9 @@ All notable changes to SkillOpt are documented here. This project adheres to
 ## [Unreleased]
 
 ### Added
+- Per-night SkillOpt-Sleep `evidence.jsonl` chains for reconstructing harvest,
+  mining, replay, reflection, and gate decisions, plus a live prompt-template
+  registry with user overrides.
 - Native SkillOpt-Sleep support for Cursor, including a local plugin command
   and skill, Cursor transcript harvesting, and an optional Cursor Agent CLI
   backend. Cursor tool-aware replay remains disabled pending live permission-

docs/sleep/README.md

@@ -42,6 +42,15 @@ experience → long-term competence).
 > temporarily disabled pending live permission-boundary validation.
 > Cursor and the model provider selected by Cursor may therefore receive
 > transcript-derived content.
+>
+> By default, each stateful night also writes a local `evidence.jsonl` under
+> the project staging tree (beside the report when one is staged); dry-runs
+> write evidence under the configured Sleep state directory. The log contains
+> best-effort-redacted, per-field-truncated copies
+> of miner, replay, judge, and reflection prompts and replies. Treat it as
+> sensitive local data and apply an appropriate retention policy. Set
+> `"evidence_log": false` to disable it; setting `"redact_secrets": false`
+> deliberately disables this defense-in-depth redaction.
 
 ## How to use it
 

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":"<rule>","anchor":"<text to replace/delete, optional>","rationale":"<why>"}].\n\n'
-            f"# Current {target}\n{cur_doc}\n"
-            f"{guard_text}"
-            f"{criteria_text}\n"
-            f"{pref_text}\n\n"
-            f"# Recurring failures\n{fail_text}"
-        )
+        from skillopt_sleep import prompts as prompt_registry
+        prompt = prompt_registry.render("reflect", {
+            "__EDIT_BUDGET__": str(edit_budget),
+            "__TARGET__": target,
+            "__CUR_DOC__": cur_doc,
+            "__GUARD__": guard_text,
+            "__CRITERIA__": criteria_text,
+            "__PREFS__": pref_text,
+            "__FAILURES__": fail_text,
+        })
         # Call with one retry: transient non-JSON replies otherwise waste a whole
         # night (the gate sees no edits and rejects). A firmer second prompt
         # recovers most of these.
+        ev = getattr(self, "evidence", None)
         arr = None
         for attempt in range(2):
             p = prompt if attempt == 0 else (
@@ -523,6 +520,11 @@ 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

skillopt_sleep/config.py

@@ -40,6 +40,14 @@
     # ── optimizer ──────────────────────────────────────────────────────────
     "backend": "mock",            # "mock" | "claude" | "codex" | "copilot" | "cursor"
     "model": "",                  # backend-specific; "" => backend default
+    # Dual-backend split (both empty => single backend above plays all roles).
+    # target = the model whose skill is deployed (runs `attempt` rollouts);
+    # optimizer = the model that mines tasks, judges rubrics, writes edits.
+    "optimizer_backend": "",
+    "optimizer_model": "",
+    "target_backend": "",
+    "target_model": "",
+    "azure_endpoint": "",         # explicit endpoint for azure/compat backends
     "gate_mode": "on",            # "on" (validation-gated) | "off" (greedy, no hard filter)
     "codex_path": "",             # "" => auto-detect the real @openai/codex binary
     "cursor_path": "",            # "" => auto-detect the Cursor Agent CLI
@@ -58,6 +66,9 @@
     "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",

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,