Skip to content

feat(sleep): local control-panel dashboard (stacked on evidence chain) #152

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Alphaxalchemy wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat(sleep): local control-panel dashboard (stacked on evidence chain) #152

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7829552
feat(sleep): per-night evidence chain and live prompt registry
Alphaxalchemy Jul 19, 2026
fd6dc42
feat(sleep): local web control panel (dashboard) for the sleep pipeline
Alphaxalchemy Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions skillopt_sleep/__main__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -513,6 +513,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":
Expand All @@ -529,6 +535,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

Expand Down
92 changes: 47 additions & 45 deletions skillopt_sleep/backend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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 -----------------------------------------------------------
Expand Down Expand Up @@ -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)
Expand All @@ -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")
Expand Down Expand Up @@ -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 (
Expand All @@ -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
Expand Down
11 changes: 11 additions & 0 deletions skillopt_sleep/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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",
Expand Down
51 changes: 50 additions & 1 deletion skillopt_sleep/consolidate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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] = []
Expand All @@ -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]
Expand All @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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]
Expand All @@ -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)
Expand All @@ -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,
Expand Down
Loading