Skip to content
119 changes: 103 additions & 16 deletions skillopt_sleep/judges.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,24 @@
* max_chars <n> — response length <= n
* min_chars <n> — response length >= n
* contains <text> — substring present (case-insensitive)
* not_contains <text> — substring absent (case-insensitive)
* no_refusal — the response is not a bare refusal/abstention
* tool_called <name> — a tool with <name> was invoked (needs a tool loop;
in single-shot replay we approximate via an
explicit "TOOL_CALL: <name>" marker the agent emits)

Ops divide into two families, and the distinction is load-bearing:

* *shape* ops (section_present, max_chars, min_chars) constrain how an answer
is formatted. They are trivially satisfiable by an optimizer — adding a
heading scores 1.0 without the answer improving at all.
* *outcome* ops (contains, not_contains, no_refusal, regex, tool_called)
constrain what the answer actually does.

A judge built only from shape ops is a formatting checker, not a grader; see
:func:`is_shape_only`. Callers should prefer an outcome-based judge (rubric or
outcome ops) so the gate cannot be won by reformatting.

A task whose judge is {"kind": "rule", "checks": [...]} passes (hard=1.0) iff
ALL checks pass; soft = fraction of checks passed. This mirrors gbrain's
all-checks-must-pass rule scoring and gives the gate a smooth signal.
Expand All @@ -35,6 +49,39 @@ def _section_present(response: str, name: str) -> bool:
return bool(label.search(response or ""))


_REFUSAL_PREFIXES = (
"cannot complete",
"i cannot",
"i can't",
"i'm unable",
"i am unable",
"unable to complete",
"sorry, i can't",
"sorry, i cannot",
"no can do",
)


def _is_refusal(response: str) -> bool:
"""Detect a bare refusal: an abstention with no substantive work reported.

A refusal that still explains what was searched and what is missing is a
useful answer, so only short responses whose opening is an abstention count.
"""
text = (response or "").strip()
if not text:
return True
# Strip leading markdown markers -- blockquote (>), list bullets (-, *),
# numbered items (1. / 1)), emphasis and headings -- BEFORE bounding the
# head, so a refusal cannot hide behind >160 marker characters.
head = re.sub(r"^(?:[>\-*_#\s]|\d+[.)])+", "", text.lower())[:160]
if not any(head.startswith(p) for p in _REFUSAL_PREFIXES):
return False
# A long response that opens with an abstention still did the work of
# explaining why; only terse dead-ends are refusals.
return len(text) < 600


def _check(op: str, arg: Any, response: str,
tools_called: List[str]) -> Tuple[bool, str]:
"""Evaluate one check.
Expand All @@ -60,6 +107,10 @@ def _check(op: str, arg: Any, response: str,
return len(r) >= int(arg), ""
if op == "contains":
return str(arg).lower() in r.lower(), ""
if op == "not_contains":
return str(arg).lower() not in r.lower(), ""
if op == "no_refusal":
return not _is_refusal(r), ""
if op == "tool_called":
name = str(arg).lower()
if any(name == t.lower() for t in tools_called):
Expand All @@ -71,9 +122,53 @@ def _check(op: str, arg: Any, response: str,


KNOWN_OPS = frozenset({
"section_present", "regex", "max_chars", "min_chars", "contains", "tool_called",
"section_present", "regex", "max_chars", "min_chars", "contains",
"not_contains", "no_refusal", "tool_called",
})

# Ops that only constrain formatting. An optimizer satisfies these by editing
# the output template, which is why a judge made solely of them is gameable.
SHAPE_OPS = frozenset({"section_present", "max_chars", "min_chars"})


def is_shape_only(judge: Any) -> bool:
"""True when every check in ``judge`` merely constrains formatting.

Such a judge cannot distinguish a better answer from a reformatted one, so
callers should prefer outcome grading (a rubric) instead of trusting it.
"""
if not isinstance(judge, dict):
return False
checks = judge.get("checks", []) or []
if not isinstance(checks, list) or not checks:
return False
ops = [c.get("op") for c in checks if isinstance(c, dict)]
if not ops or len(ops) != len(checks):
return False
return all(op in SHAPE_OPS for op in ops)


def char_bound(arg: Any) -> int:
"""Parse a ``max_chars``/``min_chars`` argument, or raise ``ValueError``.

Shared by :func:`validate_checks` and the LLM miner so the accepted shapes
cannot drift apart: a miner that emits a bound the validator later rejects
produces a tasks file that fails its own validation. ``bool`` is refused
(it is an ``int`` subclass, so ``int(True)`` would silently become 1) and a
non-integral float is refused rather than truncated.
"""
if isinstance(arg, bool):
raise ValueError("bool is not a char bound")
if isinstance(arg, int):
return arg
if isinstance(arg, float):
if not arg.is_integer():
raise ValueError("non-integral float is not a char bound")
return int(arg) # may raise OverflowError for inf/nan
if isinstance(arg, str) and re.fullmatch(r"[+-]?\d+", arg.strip()):
return int(arg.strip())
raise ValueError("not a char bound")


def validate_checks(judge: Any) -> Tuple[List[str], List[str]]:
"""Return ``(errors, warnings)`` for a rule judge's checks.
Expand Down Expand Up @@ -101,7 +196,7 @@ def validate_checks(judge: Any) -> Tuple[List[str], List[str]]:
f"check #{i} op must be a string, got {type(op).__name__}"
)
continue
if op in {"regex", "section_present", "contains", "tool_called"} and (
if op in {"regex", "section_present", "contains", "tool_called", "not_contains"} and (
arg is None or not str(arg).strip()
):
errors.append(f"check #{i} {op} needs a non-empty arg")
Expand All @@ -113,20 +208,7 @@ def validate_checks(judge: Any) -> Tuple[List[str], List[str]]:
errors.append(f"check #{i} regex does not compile ({exc}): {arg!r}")
elif op in {"max_chars", "min_chars"}:
try:
if isinstance(arg, bool):
raise ValueError
if isinstance(arg, int):
bound = arg
elif isinstance(arg, float):
if not arg.is_integer():
raise ValueError
bound = int(arg)
elif isinstance(arg, str) and re.fullmatch(
r"[+-]?\d+", arg.strip()
):
bound = int(arg.strip())
else:
raise ValueError
bound = char_bound(arg)
except (OverflowError, TypeError, ValueError):
errors.append(f"check #{i} {op} needs an integer arg, got {arg!r}")
else:
Expand All @@ -136,6 +218,11 @@ def validate_checks(judge: Any) -> Tuple[List[str], List[str]]:
warnings.append(f"check #{i} min_chars=0 always passes")
elif op not in KNOWN_OPS:
warnings.append(f"check #{i} has unknown op {op!r} — it always passes")
if not errors and is_shape_only(judge):
warnings.append(
"judge is shape-only (formatting checks); it can be satisfied by "
"reformatting rather than by a better answer"
)
return errors, warnings


Expand Down
69 changes: 54 additions & 15 deletions skillopt_sleep/llm_miner.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,11 @@

For each recurring intent it extracts:
* a clean, generalized `intent` (the reusable task, stripped of one-off specifics)
* a `rubric` (what a good answer must satisfy) -> stored as a rule judge of
`contains`/`regex`/`section_present` checks the local judge can score, OR a
free-text rubric scored by the backend's judge() when no programmatic check fits
* a `rubric` (what a good answer must satisfy). A rubric, when present, is
always stored as the reference and scored by the backend's judge() -- it
resists the reward-hacking that literal/format checks invite. Programmatic
`contains`/`regex`/`section_present`/`tool_called` checks are kept only as a
fallback reference for the intents where the miner supplies no rubric.
* a preference signal (was the user satisfied?) to weight failures

It is deliberately conservative: it only emits a task when it can name a
Expand All @@ -18,12 +20,11 @@
"""
from __future__ import annotations

import json
import re
from typing import Any, Callable, Dict, List

from skillopt_sleep import prompts as prompt_registry
Comment thread
lufen marked this conversation as resolved.
from skillopt_sleep.backend import Backend, _extract_json
from skillopt_sleep.judges import char_bound
from skillopt_sleep.mine import session_skill_hint
from skillopt_sleep.types import SessionDigest, TaskRecord

Expand All @@ -46,32 +47,70 @@ def _mk_task(d: SessionDigest, obj: Dict[str, Any], idx: int) -> TaskRecord | No
if len(intent) < 8:
return None
checks = obj.get("checks") or []
rubric = str(obj.get("rubric", "")).strip()
# A non-string rubric (e.g. a JSON null) must not become the literal "None"
# and outrank the checks; only a real string counts as a rubric.
rubric_raw = obj.get("rubric")
rubric = rubric_raw.strip() if isinstance(rubric_raw, str) else ""
satisfied = bool(obj.get("satisfied", False))

# keep only well-formed checks
# Keep only well-formed checks: the scorer runs these verbatim during
# replay, so a max_chars/min_chars with a non-integer arg (or an arg-less
# op that needs one) would crash or fail forever. Drop those here, and keep
# the accepted shapes aligned with validate_checks() so a mined tasks file
# never fails validation later (it rejects bools and negative bounds).
_needs_str_arg = {"section_present", "regex", "contains", "not_contains", "tool_called"}
# Trimming a regex would change what it matches (leading/trailing spaces are
# significant in a pattern), so only substring/tool/heading args are stripped.
_strip_arg = _needs_str_arg - {"regex"}
clean_checks = []
for c in checks:
if isinstance(c, dict) and c.get("op") in {
"section_present", "regex", "contains", "max_chars", "min_chars",
}:
clean_checks.append({"op": c["op"], "arg": c.get("arg")})
if not isinstance(c, dict):
continue
op = c.get("op")
arg = c.get("arg")
if op in _needs_str_arg:
# Store the stripped value: stray whitespace would otherwise become
# part of the required substring / tool name.
if isinstance(arg, str) and arg.strip():
clean_checks.append(
{"op": op, "arg": arg.strip() if op in _strip_arg else arg}
)
elif op in {"max_chars", "min_chars"}:
# Shared parser with validate_checks() so the two cannot drift:
# rejects bools, non-integral floats and inf/nan (OverflowError).
try:
bound = char_bound(arg)
except (OverflowError, TypeError, ValueError):
continue
if bound < 0:
continue
clean_checks.append({"op": op, "arg": bound})
elif op == "no_refusal":
clean_checks.append({"op": op, "arg": None})

import hashlib
tid = "llm_" + hashlib.sha256((d.project + intent).encode()).hexdigest()[:12]

if clean_checks:
judge = {"kind": "rule", "checks": clean_checks}
# The optimizer edits skill text that is prepended to the model's context,
# so ANY check for a literal string in the response can be satisfied by
# instructing the model to emit that string -- observed twice in practice,
# first with section_present and then with contains. Only semantic grading
# resists that, so a rubric always wins when the miner supplies one.
# Rule judges remain for imported gbrain-style benchmarks, which carry
# checks but no rubric.
if rubric:
return TaskRecord(
id=tid, project=d.project, intent=intent,
reference_kind="rule", judge={"kind": "rule", "checks": clean_checks},
reference_kind="rubric", reference=rubric,
outcome="success" if satisfied else "fail",
tags=["mined:llm"], source_sessions=[d.session_id],
skill_hint=session_skill_hint(d),
)
if rubric:
if clean_checks:
return TaskRecord(
id=tid, project=d.project, intent=intent,
reference_kind="rubric", reference=rubric,
reference_kind="rule", judge=judge,
outcome="success" if satisfied else "fail",
tags=["mined:llm"], source_sessions=[d.session_id],
skill_hint=session_skill_hint(d),
Expand Down
22 changes: 16 additions & 6 deletions skillopt_sleep/prompts.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,19 +33,29 @@
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.
GENERAL rule would help next time (correctness, completeness, tool-use,
conventions). Skip one-off or purely exploratory requests.

For each task return:
- "intent": the reusable request, generalized (no one-off specifics)
- "checks": a list of programmatic success checks a grader can run on a future
answer. Each check is one of:
{"op":"section_present","arg":"<heading text>"}
answer. Prefer checks about WHAT THE ANSWER DOES over how it is formatted:
{"op":"contains","arg":"<substring a correct answer must contain>"}
{"op":"not_contains","arg":"<substring a correct answer must NOT contain>"}
{"op":"no_refusal"}
{"op":"regex","arg":"<python regex the answer must match>"}
{"op":"contains","arg":"<substring the answer must contain>"}
{"op":"tool_called","arg":"<tool the task requires>"}
Formatting checks are available but weak, because an assistant can satisfy
them by reformatting without answering any better. Use them only alongside
an outcome check, never alone:
{"op":"section_present","arg":"<heading text>"}
{"op":"max_chars","arg":<int>}
{"op":"min_chars","arg":<int>}
Only include checks you are confident a GOOD answer must satisfy.
- "rubric": a one-sentence description of what a good answer looks like
- "rubric": a one-sentence description of what a GOOD answer achieves —
judged on substance, not on wording or layout. ALWAYS provide this. It is
the primary grader, because an assistant can satisfy any literal-string
check simply by emitting that string.
- "satisfied": true/false — did the user seem satisfied with the assistant's answer?

Return ONLY a JSON array (possibly empty). No prose.
Expand Down
17 changes: 13 additions & 4 deletions tests/test_judges.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

import pytest

from skillopt_sleep.judges import KNOWN_OPS, score_rule_judge, validate_checks
from skillopt_sleep.judges import KNOWN_OPS, SHAPE_OPS, score_rule_judge, validate_checks
from skillopt_sleep.tasks_file import load_tasks_file


Expand Down Expand Up @@ -128,8 +128,9 @@ def test_zero_min_chars_is_flagged_as_toothless(self) -> None:
{"checks": [{"op": "min_chars", "arg": 0}]}
)
self.assertEqual(errors, [])
self.assertEqual(len(warnings), 1)
self.assertIn("always passes", warnings[0])
# min_chars is also a shape op, so a lone one now draws a second
# warning; the toothless-bound warning must still be present.
self.assertTrue(any("always passes" in w for w in warnings), warnings)

def test_empty_string_operator_arguments_are_errors(self) -> None:
for op in ("regex", "section_present", "contains", "tool_called"):
Expand Down Expand Up @@ -178,7 +179,15 @@ def test_every_known_op_is_accepted(self) -> None:
for op in KNOWN_OPS:
arg = 1 if op.endswith("_chars") else "x"
errors, warnings = validate_checks({"checks": [{"op": op, "arg": arg}]})
self.assertEqual((errors, warnings), ([], []), op)
self.assertEqual(errors, [], op)
if op in SHAPE_OPS:
# A lone formatting op is still accepted, but is now flagged as
# gameable -- exactly the shape-only warning must fire.
self.assertTrue(warnings, (op, warnings))
self.assertTrue(all("shape-only" in w for w in warnings), (op, warnings))
else:
# An outcome op is real signal; it must not warn.
self.assertEqual(warnings, [], (op, warnings))


@pytest.mark.parametrize("bad_judge", [[], "", 0])
Expand Down
Loading