diff --git a/plugins/openclaw/skillopt_sleep_openclaw.py b/plugins/openclaw/skillopt_sleep_openclaw.py index 2faceaf2..92fe3cf9 100644 --- a/plugins/openclaw/skillopt_sleep_openclaw.py +++ b/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 diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py index 99e41f41..3311f59d 100644 --- a/skillopt_sleep/backend.py +++ b/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", ""): diff --git a/skillopt_sleep/judges.py b/skillopt_sleep/judges.py index bdcf982b..1ac68a03 100644 --- a/skillopt_sleep/judges.py +++ b/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 diff --git a/skillopt_sleep/llm_miner.py b/skillopt_sleep/llm_miner.py index d3134426..e52028a2 100644 --- a/skillopt_sleep/llm_miner.py +++ b/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] diff --git a/skillopt_sleep/prompts.py b/skillopt_sleep/prompts.py index 9d896e15..3c297e5c 100644 --- a/skillopt_sleep/prompts.py +++ b/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":""} - {"op":"not_contains","arg":""} - {"op":"no_refusal"} - {"op":"regex","arg":""} - {"op":"tool_called","arg":""} + 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":"","description":""} + {"op":"not_contains","arg":"","description":""} + {"op":"no_refusal","description":"Complete the requested task instead of refusing"} + {"op":"regex","arg":"","description":""} + {"op":"tool_called","arg":"","description":""} 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' " diff --git a/skillopt_sleep/replay.py b/skillopt_sleep/replay.py index 1502a71c..c60b0abb 100644 --- a/skillopt_sleep/replay.py +++ b/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 ""), ) diff --git a/skillopt_sleep/rollout.py b/skillopt_sleep/rollout.py index f889333b..d864c02f 100644 --- a/skillopt_sleep/rollout.py +++ b/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" diff --git a/skillopt_sleep/slow_update.py b/skillopt_sleep/slow_update.py index 72627853..ccb021bc 100644 --- a/skillopt_sleep/slow_update.py +++ b/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_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] diff --git a/skillopt_sleep/types.py b/skillopt_sleep/types.py index b40578a6..150a4f58 100644 --- a/skillopt_sleep/types.py +++ b/skillopt_sleep/types.py @@ -110,6 +110,11 @@ class ReplayResult: tools_called: List[str] = field(default_factory=list) tokens: int = 0 # approx tokens this rollout cost (for token objective) latency_ms: float = 0.0 # wall-clock for this rollout (for latency objective) + # Semantic, task-level feedback that may be shown to the skill optimizer. + # ``fail_reason`` and ``judge_rationale`` remain the verbatim verifier + # evidence for audit/debugging; they may contain regexes or other grader + # implementation details and must not be used as learning context. + optimizer_feedback: str = "" def to_dict(self) -> Dict[str, Any]: return asdict(self) diff --git a/tests/test_judges.py b/tests/test_judges.py index a68946fb..c11aa439 100644 --- a/tests/test_judges.py +++ b/tests/test_judges.py @@ -9,7 +9,13 @@ import pytest -from skillopt_sleep.judges import KNOWN_OPS, SHAPE_OPS, score_rule_judge, validate_checks +from skillopt_sleep.judges import ( + KNOWN_OPS, + SHAPE_OPS, + score_rule_judge, + score_rule_judge_with_feedback, + validate_checks, +) from skillopt_sleep.tasks_file import load_tasks_file @@ -113,6 +119,56 @@ def test_empty_checks_score_zero(self) -> None: self.assertEqual(score_rule_judge({"kind": "rule", "checks": []}, "x")[:2], (0.0, 0.0)) +class TestOptimizerFeedbackSeparation(unittest.TestCase): + def test_regex_remains_in_audit_rationale_but_not_optimizer_feedback(self) -> None: + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + hard, soft, rationale, feedback = score_rule_judge_with_feedback( + {"kind": "rule", "checks": [{"op": "regex", "arg": pattern}]}, + "No route declaration here.", + ) + + self.assertEqual((hard, soft), (0.0, 0.0)) + self.assertIn(pattern, rationale) + self.assertNotIn(pattern, feedback) + self.assertNotIn("regex", feedback.lower()) + self.assertIn("private", feedback.lower()) + + def test_plain_language_description_is_the_optimizer_signal(self) -> None: + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + _hard, _soft, rationale, feedback = score_rule_judge_with_feedback( + { + "kind": "rule", + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ], + }, + "No route declaration here.", + ) + + self.assertIn(pattern, rationale) + self.assertEqual(feedback, description) + + def test_regex_copied_into_description_fails_closed(self) -> None: + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + _hard, _soft, _rationale, feedback = score_rule_judge_with_feedback( + { + "kind": "rule", + "checks": [ + { + "op": "regex", + "arg": pattern, + "description": f"Make the answer match {pattern}", + } + ], + }, + "No route declaration here.", + ) + + self.assertNotIn(pattern, feedback) + self.assertIn("private", feedback.lower()) + + class TestMalformedRegexIsDistinguishable(unittest.TestCase): """A pattern Python cannot parse must not read like a plain miss.""" diff --git a/tests/test_outcome_judges.py b/tests/test_outcome_judges.py index e9ec67e2..330c8d53 100644 --- a/tests/test_outcome_judges.py +++ b/tests/test_outcome_judges.py @@ -356,6 +356,28 @@ def test_miner_preserves_regex_whitespace() -> None: assert errors == [] +def test_miner_preserves_plain_language_check_description() -> None: + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + task = _mk_task( + _digest(), + { + "intent": "route a recurring consultation request", + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ], + "rubric": "", + "satisfied": False, + }, + 0, + ) + + assert task is not None + assert task.judge["checks"] == [ + {"op": "regex", "arg": pattern, "description": description} + ] + + def test_mined_checks_always_pass_validate_checks() -> None: # The miner must never emit a judge that validate_checks() later rejects: diff --git a/tests/test_sleep_engine.py b/tests/test_sleep_engine.py index 3f975361..2e6f4692 100644 --- a/tests/test_sleep_engine.py +++ b/tests/test_sleep_engine.py @@ -1033,6 +1033,201 @@ def _call(self, prompt, *, max_tokens=1024): [], "skill", "", edit_budget=2, evolve_skill=True, evolve_memory=False) self.assertIn("British English", captured["prompt"]) + def test_reflect_does_not_receive_raw_verifier_syntax(self): + from skillopt_sleep.backend import CliBackend + from skillopt_sleep.types import ReplayResult + + captured = {} + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + + class CapBackend(CliBackend): + name = "cap" + + def _call(self, prompt, *, max_tokens=1024): + captured["prompt"] = prompt + return "[]" + + task = TaskRecord( + id="t", + project="/p", + intent="Route a consultation request", + reference_kind="rule", + judge={ + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ] + }, + ) + # No optimizer_feedback on purpose: legacy/deserialized results must be + # projected safely from the task instead of falling back to fail_reason. + result = ReplayResult( + id="t", + hard=0.0, + response="No route declaration here.", + fail_reason=f"failed: regex={pattern}", + ) + + CapBackend().reflect( + [(task, result)], + [], + "skill", + "", + edit_budget=2, + evolve_skill=True, + evolve_memory=False, + ) + + self.assertIn(description, captured["prompt"]) + self.assertNotIn(pattern, captured["prompt"]) + self.assertNotIn("regex=", captured["prompt"]) + + def test_legacy_non_rule_feedback_fails_closed(self): + from skillopt_sleep.backend import _optimizer_feedback + from skillopt_sleep.types import ReplayResult + + raw_evidence = "judge implementation: private-evaluator-expression" + task = TaskRecord( + id="t", + project="/p", + intent="Answer the request", + reference_kind="rubric", + reference="Give a helpful answer.", + ) + result = ReplayResult( + id="t", + hard=0.0, + fail_reason=raw_evidence, + judge_rationale=raw_evidence, + ) + + feedback = _optimizer_feedback(task, result) + + self.assertNotIn(raw_evidence, feedback) + self.assertIn("did not satisfy", feedback) + + def test_supplied_rule_feedback_is_recomputed_from_safe_description(self): + from skillopt_sleep.backend import _optimizer_feedback + from skillopt_sleep.types import ReplayResult + + pattern = r"(?im)^\\s*SKILL:\\s*jyoti-prashna-util\\s*$" + description = "Route this class of request to the consultation utility." + task = TaskRecord( + id="t", + project="/p", + intent="Route a consultation request", + reference_kind="rule", + judge={ + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ] + }, + ) + result = ReplayResult( + id="t", + hard=0.0, + response="No route declaration here.", + optimizer_feedback=f"unsafe regex={pattern}", + ) + + feedback = _optimizer_feedback(task, result) + + self.assertEqual(feedback, description) + self.assertNotIn(pattern, feedback) + + def test_replay_non_rule_feedback_is_generic_even_when_rationale_is_raw(self): + from skillopt_sleep.backend import Backend + from skillopt_sleep.replay import replay_one + + pattern = r"private-check-expression" + + class StubBackend(Backend): + name = "stub" + + def attempt(self, task, skill, memory, sample_id=0): + return "bad" + + def judge(self, task, response): + return 0.0, 0.0, f"judge implementation: {pattern}" + + task = TaskRecord( + id="t", + project="/p", + intent="Answer the request", + reference_kind="rubric", + reference="Give a helpful answer.", + ) + + result = replay_one(StubBackend(), task, "", "") + + self.assertIn(pattern, result.judge_rationale) + self.assertNotIn(pattern, result.optimizer_feedback) + self.assertIn("did not satisfy", result.optimizer_feedback) + + def test_openclaw_reflect_uses_only_optimizer_feedback(self): + import importlib.util + from pathlib import Path + + from skillopt_sleep.types import ReplayResult + + backend_path = ( + Path(__file__).resolve().parents[1] + / "plugins" + / "openclaw" + / "skillopt_sleep_openclaw.py" + ) + spec = importlib.util.spec_from_file_location( + "skillopt_sleep_openclaw_feedback_test", backend_path + ) + self.assertIsNotNone(spec) + self.assertIsNotNone(spec.loader) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + task = TaskRecord( + id="t", + project="/p", + intent="Route a consultation request", + reference_kind="rule", + judge={ + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ] + }, + ) + failure = ReplayResult( + id="t", + hard=0.0, + response="No route declaration here.", + fail_reason=f"failed: regex={pattern}", + judge_rationale=f"failed: regex={pattern}", + ) + success = ReplayResult( + id="t", + hard=1.0, + response="good", + judge_rationale=f"all checks passed: regex={pattern}", + ) + + with mock.patch.object(module, "_chat", return_value='{"edits": []}') as chat: + module.OpenClawDeepSeekBackend().reflect( + [(task, failure)], + [(task, success)], + "skill", + "", + edit_budget=2, + evolve_skill=True, + evolve_memory=False, + ) + + messages = chat.call_args.args[0] + optimizer_prompt = "\n".join(message["content"] for message in messages) + self.assertIn(description, optimizer_prompt) + self.assertNotIn(pattern, optimizer_prompt) + self.assertNotIn("regex=", optimizer_prompt) + def test_reflect_records_last_raw(self): # the optimizer's raw reply must be retained so a no-edits night is # diagnosable (empty/non-JSON reflect vs genuinely no failures). @@ -1060,6 +1255,33 @@ def test_replay_records_cost(self): self.assertGreater(r.tokens, 0) self.assertGreaterEqual(r.latency_ms, 0.0) + def test_replay_keeps_raw_evidence_separate_from_optimizer_feedback(self): + from skillopt_sleep.backend import MockBackend + from skillopt_sleep.replay import replay_one + + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + task = TaskRecord( + id="t", + project="/p", + intent="Route a consultation request", + reference_kind="rule", + judge={ + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ] + }, + ) + + result = replay_one(MockBackend(), task, "", "") + serialized = result.to_dict() + + self.assertIn(pattern, result.fail_reason) + self.assertIn(pattern, result.judge_rationale) + self.assertEqual(result.optimizer_feedback, description) + self.assertIn(pattern, serialized["fail_reason"]) + self.assertNotIn(pattern, serialized["optimizer_feedback"]) + class TestCodexBackend(unittest.TestCase): def test_codex_cli_backend_runs_exec_in_project_dir(self): @@ -1273,6 +1495,52 @@ def _call(self, prompt, *, max_tokens=1024): self.assertEqual(len(edits), 1) self.assertIn("good thing", edits[0].content) + def test_contrastive_reflect_hides_raw_verifier_syntax(self): + from skillopt_sleep.backend import Backend + from skillopt_sleep.rollout import RolloutSet, contrastive_reflect + from skillopt_sleep.types import ReplayResult + + captured = {} + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + + class StubBackend(Backend): + name = "stub" + + def _call(self, prompt, *, max_tokens=1024): + captured["prompt"] = prompt + return "[]" + + task = TaskRecord( + id="t", + project="/p", + intent="route a consultation request", + reference_kind="rule", + judge={ + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ] + }, + ) + rs = RolloutSet( + task=task, + attempts=[ + ReplayResult(id="t", hard=1.0, response="good"), + ReplayResult( + id="t", + hard=0.0, + response="bad", + fail_reason=f"failed: regex={pattern}", + ), + ], + ) + + contrastive_reflect(StubBackend(), [rs], "skill", "") + + self.assertIn(description, captured["prompt"]) + self.assertNotIn(pattern, captured["prompt"]) + self.assertNotIn("regex=", captured["prompt"]) + class TestSlowUpdate(unittest.TestCase): def test_protected_field_roundtrip(self): @@ -1319,6 +1587,58 @@ def _call(self, prompt, *, max_tokens=1024): prev_pairs=prev2, curr_pairs=curr2) self.assertIn("keep doing X", out2) + def test_slow_update_hides_raw_verifier_syntax(self): + from skillopt_sleep.backend import Backend + from skillopt_sleep.slow_update import run_slow_update + from skillopt_sleep.types import ReplayResult + + captured = {} + pattern = r"(?im)^\s*SKILL:\s*jyoti-prashna-util\s*$" + description = "Route this class of request to the consultation utility." + + class StubBackend(Backend): + name = "stub" + + def _call(self, prompt, *, max_tokens=1024): + captured["prompt"] = prompt + return '{"guidance": "keep routing consultation requests"}' + + task = TaskRecord( + id="t", + project="/p", + intent="route a consultation request", + reference_kind="rule", + judge={ + "checks": [ + {"op": "regex", "arg": pattern, "description": description} + ] + }, + ) + previous = [(task, ReplayResult(id="t", hard=1.0))] + current = [ + ( + task, + ReplayResult( + id="t", + hard=0.0, + response="bad", + fail_reason=f"failed: regex={pattern}", + ), + ) + ] + + run_slow_update( + StubBackend(), + prev_skill="s0", + curr_skill="s1", + prev_pairs=previous, + curr_pairs=current, + ) + + self.assertIn(description, captured["prompt"]) + self.assertNotIn(pattern, captured["prompt"]) + self.assertNotIn("regex=", captured["prompt"]) + class TestToolLoop(unittest.TestCase): def test_tool_called_judge_via_replay(self):