fix(sleep): separate optimizer feedback from judge evidence

plugins/openclaw/skillopt_sleep_openclaw.py

@@ -12,13 +12,11 @@
 import json
 import os
 import re
-import subprocess
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Dict, List, Tuple
 
-from skillopt_sleep.backend import Backend, _normalize, exact_score
+from skillopt_sleep.backend import Backend, _optimizer_feedback, exact_score
 from skillopt_sleep.types import EditRecord, ReplayResult, TaskRecord
 
-
 # ── DeepSeek + Ollama OpenAI-compatible API client (curl-based, no extra deps) ──
 
 
@@ -194,12 +192,13 @@ def reflect(
     ) -> List[EditRecord]:
         # Compact digest of failures + successes
         fail_digest = "\n".join(
-            f"- TASK: {t.intent[:200]}\n  RESPONSE: {r.response[:300]}\n  WHY FAIL: {r.judge_rationale or r.fail_reason or 'unknown'}\n  REFERENCE: {t.reference[:200]}"
+            f"- TASK: {t.intent[:200]}\n  RESPONSE: {r.response[:300]}\n"
+            f"  ACTIONABLE FEEDBACK: {_optimizer_feedback(t, r)[:240]}\n"
+            f"  REFERENCE: {t.reference[:200]}"
             for t, r in failures[:5]
         ) or "(none)"
         succ_digest = "\n".join(
-            f"- TASK: {t.intent[:150]} -> OK ({r.judge_rationale or 'high score'})"
-            for t, r in successes[:3]
+            f"- TASK: {t.intent[:150]} -> OK" for t, _r in successes[:3]
         ) or "(none)"
 
         rubric_text = ""
@@ -216,6 +215,7 @@ def reflect(
             "that, if applied, would help future agents do better on the failed tasks. "
             "NEVER propose adding new sections wholesale. NEVER delete entire sections. "
             "Edit primitives: ADD (append a step/rule at end), DELETE (remove a specific line by exact match), REPLACE (swap a specific line for another by exact match). "
+            "Use only the actionable semantic feedback; never quote or reconstruct private verifier syntax such as regexes or check source. "
             "If you cannot identify a clear, minimal improvement, return an empty list."
         )
         usr = f"""## CURRENT SKILL

skillopt_sleep/backend.py

@@ -124,6 +124,31 @@ def keyword_soft_score(reference: str, response: str) -> float:
     return hit / len(set(ref_tokens))
 
 
+def _optimizer_feedback(task: TaskRecord, result: ReplayResult) -> str:
+    """Return learning-safe feedback while retaining raw evidence elsewhere.
+
+    Recompute semantic rule-judge feedback for legacy/deserialized results that
+    predate ``ReplayResult.optimizer_feedback``. This makes every reflection
+    path safe even when callers construct ReplayResult objects themselves.
+    """
+    if task.reference_kind == "rule" and task.judge:
+        from skillopt_sleep.judges import score_rule_judge_with_feedback
+
+        _hard, _soft, _rationale, feedback = score_rule_judge_with_feedback(
+            task.judge,
+            getattr(result, "response", ""),
+            getattr(result, "tools_called", []),
+        )
+        return feedback
+    feedback = getattr(result, "optimizer_feedback", "")
+    if isinstance(feedback, str) and feedback:
+        return feedback
+    # Legacy results for non-rule judges cannot be safely projected: their raw
+    # rationale may mix semantic guidance with grader internals. Fail closed
+    # rather than feeding that evidence back into an optimizer prompt.
+    return "The response did not satisfy the task's evaluation criteria."
+
+
 # ── Mock backend (deterministic, no API) ──────────────────────────────────────
 
 class MockBackend(Backend):
@@ -446,49 +471,25 @@ def reflect(
         target = "skill" if evolve_skill else "memory"
         cur_doc = (skill if target == "skill" else memory) or "(empty)"
         fail_text = "\n".join(
-            f"- wanted: {t.intent[:160]}\n  got: {r.response[:160]}\n  why-wrong: {r.fail_reason[:160]}"
+            f"- wanted: {t.intent[:160]}\n  got: {r.response[:160]}\n"
+            f"  actionable-feedback: {_optimizer_feedback(t, r)[:240]}"
             for t, r in failures[:8]
         )
-        # Aggregate the most common failing criteria across all failures so the
-        # optimizer is told *exactly what the scorer rewards* — gbrain's lesson:
-        # the optimizer kept proposing reasonable-but-wrong edits until it could
-        # see the success criteria.
+        # Aggregate semantic requirements, never raw verifier syntax. Exact
+        # regexes/check expressions stay in ReplayResult.fail_reason and the
+        # evidence log for audit, but are not instructions to the optimizer.
         from collections import Counter
         crit = Counter()
-        for _t, r in failures:
-            fr = r.fail_reason or ""
-            if fr.startswith("failed:"):
-                for part in fr[len("failed:"):].split(","):
-                    part = part.strip()
-                    if part:
-                        crit[part] += 1
-
-        def _explain(c: str) -> str:
-            # translate an "op=arg" criterion into a plain-English requirement
-            if "=" in c:
-                op, _, arg = c.partition("=")
-                op = op.strip(); arg = arg.strip()
-                if op == "max_chars":
-                    return f"the ENTIRE response must be at most {arg} characters long"
-                if op == "min_chars":
-                    return f"the response must be at least {arg} characters long"
-                if op == "section_present":
-                    return f"the response must contain a section/heading titled '{arg}'"
-                if op == "section_contains":
-                    return f"a markdown heading must contain the text '{arg}'"
-                if op == "regex":
-                    return f"the response must match the pattern /{arg}/ (e.g. include that label)"
-                if op == "contains":
-                    return f"the response must contain the text '{arg}'"
-                if op == "tool_called":
-                    return f"the agent must actually call the '{arg}' tool"
-            return c
+        for t, r in failures:
+            feedback = _optimizer_feedback(t, r).strip()
+            if feedback:
+                crit[feedback] += 1
 
         criteria_text = ""
         if crit:
             criteria_text = (
-                "\n# Exact criteria the outputs are FAILING (fix these directly)\n"
-                + "\n".join(f"- {_explain(c)}  [{c}, failed {n}x]" for c, n in crit.most_common())
+                "\n# Actionable semantic feedback (fix the behavior, not the verifier)\n"
+                + "\n".join(f"- {c} (observed {n}x)" for c, n in crit.most_common())
             )
         pref_text = ""
         if getattr(self, "preferences", ""):

skillopt_sleep/judges.py

@@ -250,18 +250,71 @@ def validate_checks(judge: Any) -> Tuple[List[str], List[str]]:
     return errors, warnings
 
 
-def score_rule_judge(
+def _semantic_failure(check: Dict[str, Any], problem: str = "") -> str:
+    """Describe an unmet check without exposing its implementation syntax.
+
+    Rule-judge details are evidence, not instructions for the optimizer. In
+    particular, copying a regex into reflection lets an evolving skill learn
+    the verifier rather than the intended behavior. An operator-supplied
+    plain-language description is the strongest signal; conservative built-in
+    descriptions are used for legacy checks that do not carry one.
+    """
+    if problem:
+        return (
+            "The evaluator configuration is invalid; do not change the skill "
+            "based on this result."
+        )
+
+    op = str(check.get("op", ""))
+    arg = check.get("arg")
+    description = check.get("description")
+    if isinstance(description, str) and description.strip():
+        cleaned = description.strip()
+        # A generated or hand-authored description may accidentally copy the
+        # regex it is meant to explain. Fail closed instead of laundering the
+        # verifier implementation into the optimizer channel.
+        if op != "regex" or not isinstance(arg, str) or arg not in cleaned:
+            return cleaned
+
+    if op == "max_chars":
+        return f"Keep the entire response at or below {arg} characters."
+    if op == "min_chars":
+        return f"Provide at least {arg} characters of substantive response."
+    if op == "section_present":
+        return f"Include a section or heading titled {arg!r}."
+    if op == "section_contains":
+        return f"Include {arg!r} in an ATX markdown heading."
+    if op == "contains":
+        return f"The response must include the required concept or phrase {arg!r}."
+    if op == "not_contains":
+        return f"The response included prohibited content related to {arg!r}."
+    if op == "no_refusal":
+        return "Complete the task instead of refusing it."
+    if op == "tool_called":
+        return f"Actually call the {arg!r} tool while completing the task."
+    if op == "regex":
+        return "The response did not satisfy a private content or format requirement."
+    return "The response did not satisfy one of the task requirements."
+
+
+def score_rule_judge_with_feedback(
     judge: Dict[str, Any],
     response: str,
     tools_called: List[str] | None = None,
-) -> Tuple[float, float, str]:
-    """Return (hard, soft, rationale) for a gbrain-style rule judge."""
+) -> Tuple[float, float, str, str]:
+    """Return scores, audit rationale, and optimizer-safe feedback."""
     checks = (judge or {}).get("checks", []) or []
     if not checks:
-        return 0.0, 0.0, "no checks"
+        return (
+            0.0,
+            0.0,
+            "no checks",
+            "The evaluator has no checks; do not change the skill based on this result.",
+        )
     tools_called = tools_called or []
     passed = 0
     failed_desc: List[str] = []
+    semantic_failures: List[str] = []
     for c in checks:
         ok, problem = _check(c.get("op", ""), c.get("arg"), response, tools_called)
         if ok:
@@ -271,7 +324,21 @@ def score_rule_judge(
             if problem:
                 desc += f" [{problem}]"
             failed_desc.append(desc)
+            semantic_failures.append(_semantic_failure(c, problem))
     soft = passed / len(checks)
     hard = 1.0 if passed == len(checks) else 0.0
     rationale = "all checks passed" if hard else "failed: " + ", ".join(failed_desc)
+    feedback = " ".join(dict.fromkeys(semantic_failures))
+    return hard, soft, rationale, feedback
+
+
+def score_rule_judge(
+    judge: Dict[str, Any],
+    response: str,
+    tools_called: List[str] | None = None,
+) -> Tuple[float, float, str]:
+    """Return the backward-compatible (hard, soft, rationale) tuple."""
+    hard, soft, rationale, _feedback = score_rule_judge_with_feedback(
+        judge, response, tools_called
+    )
     return hard, soft, rationale

skillopt_sleep/llm_miner.py

@@ -72,13 +72,25 @@ def _mk_task(d: SessionDigest, obj: Dict[str, Any], idx: int) -> TaskRecord | No
             continue
         op = c.get("op")
         arg = c.get("arg")
+        description_raw = c.get("description")
+        description = (
+            description_raw.strip()
+            if isinstance(description_raw, str) and description_raw.strip()
+            else ""
+        )
+
+        def _with_description(check: Dict[str, Any]) -> Dict[str, Any]:
+            if description:
+                check["description"] = description
+            return check
+
         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(
+                clean_checks.append(_with_description(
                     {"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).
@@ -88,9 +100,9 @@ def _mk_task(d: SessionDigest, obj: Dict[str, Any], idx: int) -> TaskRecord | No
                 continue
             if bound < 0:
                 continue
-            clean_checks.append({"op": op, "arg": bound})
+            clean_checks.append(_with_description({"op": op, "arg": bound}))
         elif op == "no_refusal":
-            clean_checks.append({"op": op, "arg": None})
+            clean_checks.append(_with_description({"op": op, "arg": None}))
 
     import hashlib
     tid = "llm_" + hashlib.sha256((d.project + intent).encode()).hexdigest()[:12]

skillopt_sleep/prompts.py

@@ -39,12 +39,14 @@
 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. 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":"tool_called","arg":"<tool the task requires>"}
+     answer. Give every check a plain-language "description" of the intended
+     behavior; it must not repeat regex or grader syntax. Prefer checks about
+     WHAT THE ANSWER DOES over how it is formatted:
+        {"op":"contains","arg":"<required substring>","description":"<semantic requirement>"}
+        {"op":"not_contains","arg":"<prohibited substring>","description":"<semantic requirement>"}
+        {"op":"no_refusal","description":"Complete the requested task instead of refusing"}
+        {"op":"regex","arg":"<python regex>","description":"<meaning of the pattern, without its syntax>"}
+        {"op":"tool_called","arg":"<required tool>","description":"<why the tool must be used>"}
      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:
@@ -91,12 +93,12 @@
     "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"
+    "answer to a single task). Use the actionable semantic feedback to fix "
+    "user-visible behavior, not to imitate the evaluator.\n"
+    "BE CONCRETE about public task requirements such as thresholds and section "
+    "names, but NEVER quote or reconstruct private verifier implementation "
+    "details such as regexes, check expressions, or judge source text. Vague "
+    "rules do not change behavior, while verifier-specific rules overfit.\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' "

skillopt_sleep/replay.py

@@ -49,10 +49,20 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
 
     # rule judges may need the detected tool calls; score locally when possible
     if task.reference_kind == "rule" and task.judge:
-        from skillopt_sleep.judges import score_rule_judge
-        hard, soft, rationale = score_rule_judge(task.judge, response, tools_called)
+        from skillopt_sleep.judges import score_rule_judge_with_feedback
+        hard, soft, rationale, optimizer_feedback = score_rule_judge_with_feedback(
+            task.judge, response, tools_called
+        )
     else:
         hard, soft, rationale = backend.judge(task, response)
+        # Backend judge rationales are audit evidence and may contain provider
+        # or grader implementation details. Keep the optimizer channel generic
+        # for non-rule judges unless a future typed safe-feedback API exists.
+        optimizer_feedback = (
+            "The response did not satisfy the task's evaluation criteria."
+            if hard < 1.0
+            else ""
+        )
 
     ev = getattr(backend, "evidence", None)
     if ev is not None:
@@ -77,6 +87,7 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
         tools_called=tools_called,
         tokens=int(tokens),
         latency_ms=round(latency_ms, 1),
+        optimizer_feedback=(optimizer_feedback if hard < 1.0 else ""),
     )
 
 

skillopt_sleep/rollout.py

@@ -127,6 +127,8 @@ def contrastive_reflect(
     if not informative:
         return []
 
+    from skillopt_sleep.backend import _optimizer_feedback
+
     blocks = []
     for _spread, rs, best, worst, best_score, worst_score in informative:
         blocks.append(
@@ -135,7 +137,7 @@ def contrastive_reflect(
             f"hard {best.hard:.3f}, soft {best.soft:.3f}): {best.response[:200]}\n"
             f"- BAD  attempt ({gate_metric} score {worst_score:.3f}; "
             f"hard {worst.hard:.3f}, soft {worst.soft:.3f}): {worst.response[:200]}\n"
-            f"  (bad failed: {worst.fail_reason[:100]})"
+            f"  (actionable feedback: {_optimizer_feedback(rs.task, worst)[:160]})"
         )
     # the output contract the proposed rules must not violate (same guardrail the
     # single-shot reflect uses — prevents harness-violating rules like "return VBA"
@@ -148,8 +150,9 @@ def contrastive_reflect(
         "others under the gate objective. Identify what the GOOD attempts did that "
         "the BAD ones did not, "
         f"and propose at most {edit_budget} SHORT, GENERAL, reusable rules for the "
-        f"{target} that would make the good behavior reliable every time. Quote "
-        "concrete thresholds/formats verbatim; do not paraphrase vaguely. "
+        f"{target} that would make the good behavior reliable every time. Be "
+        "concrete about public task requirements, but never quote, reconstruct, "
+        "or optimize for private verifier syntax such as regexes or check source. "
         "Every rule MUST obey the task output contract (if shown) — never propose "
         "a rule that changes the required output format/language or tells the agent "
         "to ask the user a question; such a rule scores ZERO.\n"

skillopt_sleep/slow_update.py

@@ -20,13 +20,11 @@
 """
 from __future__ import annotations
 
-import re
 from typing import List, Optional, Tuple
 
-from skillopt_sleep.backend import Backend, _extract_json
+from skillopt_sleep.backend import Backend, _extract_json, _optimizer_feedback
 from skillopt_sleep.types import ReplayResult, TaskRecord
 
-
 SLOW_UPDATE_START = "<!-- SLOW_UPDATE_START -->"
 SLOW_UPDATE_END = "<!-- SLOW_UPDATE_END -->"
 
@@ -95,7 +93,8 @@ def _summarize_pairs(
             cat = "persistent_fail"
         counts[cat] += 1
         if cat in ("regressed", "persistent_fail") and len(lines) < 8:
-            lines.append(f"- [{cat}] {t.intent[:120]} (why: {r.fail_reason[:80]})")
+            feedback = _optimizer_feedback(t, r)
+            lines.append(f"- [{cat}] {t.intent[:120]} (why: {feedback[:160]})")
     head = ", ".join(f"{k}={v}" for k, v in counts.items())
     return head + ("\n" + "\n".join(lines) if lines else ""), counts  # type: ignore[return-value]