# SPDX-License-Identifier: BSD-3-Clause """Run the layers over one submission and assemble the verdict. The weights follow the rubric Narreddi's write-up converged on after the first one, which had nine signals, plateaued: a binary compile-and-uses-brush gate at 0.05, a binary length check at 0.05, HPSv3 at 0.30, and the pairwise judge at 0.60. Two deliberate differences from that, both of which change what the numbers mean. **HPSv3's slot is filled by an absolute mark from the judge model.** HPSv3 is a 7B preference model trained on 1.17M human comparisons, and it carries 0.30 of their reward. Its *role* is what matters: it scores a painting on its own rather than against a reference, so a policy has something to climb before it can beat anything in the pool. Leaving that slot empty cost the first four runs, and filling it with a coverage term cost the next few: coverage knows how much paint landed and nothing about whether the painting is any good, so once the policy painted a plausible amount the term went flat while the pairwise judge sat at exactly zero for twenty-four rollouts in a row. See [`~envs.watercolour_env.server.quality`] for the replacement and for the measured proof that it grades where the comparison had given up. **The gate stays absolute for the dishonest cases.** Awarding 0.05 for compiling is their structure and it is worth keeping, because it is the only term a policy can earn on day one and so the only source of reward variance before the judge engages. But a sketch that loaded somebody else's painting or wrote the answer as text scores zero outright rather than collecting the compile credit. The gate runs first and is free. Only what clears it reaches the judge, and the render is carried across so the painting is produced exactly once per episode. """ from __future__ import annotations import asyncio import os from dataclasses import dataclass from typing import Any try: from .gate import GateResult, run_gate from .pairwise_judge import JudgeReport, PairwiseJudge from .quality import QualityReport, QualityScorer from .render import SketchRenderer from .tasks import Task except ImportError: from server.gate import GateResult, run_gate from server.pairwise_judge import JudgeReport, PairwiseJudge from server.quality import QualityReport, QualityScorer from server.render import SketchRenderer from server.tasks import Task # Defaults are the write-up's final rubric, kept exactly. They are read from the # environment because the split between the pairwise judge and the preference model # is the one thing worth varying: three runs of this environment differ only in # these four numbers, and hard-coding them meant a forked copy of the whole server # per combination. GATE_WEIGHT = float(os.environ.get("WATERCOLOUR_GATE_WEIGHT", 0.05)) LENGTH_WEIGHT = float(os.environ.get("WATERCOLOUR_LENGTH_WEIGHT", 0.05)) JUDGE_WEIGHT = float(os.environ.get("WATERCOLOUR_JUDGE_WEIGHT", 0.60)) # The slot HPSv3 holds in their rubric, at their weight. See # [`~envs.watercolour_env.server.quality`] for why an absolute mark from the judge # model stands in for the preference model itself, and for the numbers that say the # stand-in discriminates where the pairwise term had stopped. QUALITY_WEIGHT = float(os.environ.get("WATERCOLOUR_QUALITY_WEIGHT", 0.30)) # A ramp, not a band, and the change is deliberate. Their write-up describes "a # code length ramp targeting around 3,000 tokens" and sketches that "compressed # from 13,500 tokens to under 2,000", so elaboration is something their reward # pulls towards. The band this replaces returned one for anything between 150 and # 1200 tokens, and measured output sits inside it: a 4B writes 570 to 1256 tokens # and a VLM 700 to 1300. A term that is one for every rollout contributes nothing # to a GRPO group, so the only signal about elaboration in the whole rubric was # doing no work. # # Below the floor a sketch is degenerate and scores zero. From there it climbs to # full credit at the target, so writing more elaborate code always pays a little. # Past the runaway cap it scores zero again: their 13,500-token sketches were a # problem to be compressed, not a goal. MIN_LENGTH_TOKENS = 150 TARGET_LENGTH_TOKENS = 3000 RUNAWAY_LENGTH_TOKENS = 6000 def length_score(source: str) -> float: """Return how close a sketch's length is to the target elaboration. Args: source (`str`): The extracted sketch source. Returns: `float`: Zero below the floor and above the runaway cap, climbing linearly to one at the target in between. Examples: ```python >>> length_score("x" * 4 * 600) # what a 4B writes today 0.16 >>> length_score("x" * 4 * 3000) # the target 1.0 ``` """ tokens = len(source) / 4 if tokens < MIN_LENGTH_TOKENS or tokens > RUNAWAY_LENGTH_TOKENS: return 0.0 if tokens >= TARGET_LENGTH_TOKENS: return 1.0 return (tokens - MIN_LENGTH_TOKENS) / (TARGET_LENGTH_TOKENS - MIN_LENGTH_TOKENS) @dataclass(frozen=True) class Evaluation: """Everything known about one submission. Attributes: task ([`Task`]): The request the submission was answering. gate ([`GateResult`]): Verdict of the admission check. judge ([`JudgeReport`] or `None`): The comparative verdict, `None` when the gate rejected the submission or no judge was configured. judge_enabled (`bool`): Whether a judge was configured for this run, regardless of whether it managed to answer. quality ([`QualityReport`], *optional*): The absolute mark, `None` when the gate rejected the submission or no scorer was configured. """ task: Task gate: GateResult judge: JudgeReport | None = None judge_enabled: bool = False quality: QualityReport | None = None @property def gate_passed(self) -> bool: """`bool`: Whether the submission cleared the admission check.""" return self.gate.passed @property def judge_score(self) -> float: """`float`: The comparative score, zero when nothing was judged.""" return self.judge.score if self.judge is not None else 0.0 @property def render_unavailable(self) -> bool: """`bool`: Whether the browser failed rather than the sketch.""" return self.gate.render_unavailable @property def judged(self) -> bool: """`bool`: Whether a judge verdict was actually obtained.""" return self.judge is not None and self.judge.available @property def quality_score(self) -> float: """`float`: The absolute mark in [0, 1], zero when nothing was marked.""" return self.quality.score if self.quality is not None else 0.0 @property def quality_scored(self) -> bool: """`bool`: Whether an absolute mark was actually obtained.""" return self.quality is not None and self.quality.available @property def length_score(self) -> float: """`float`: One if the sketch's length is in the accepted band.""" if self.gate.source is None: return 0.0 return length_score(self.gate.source.source) @property def reward(self) -> float: """`float`: The episode reward. A weighted sum of the components that survive, and the components are what matter more than the total. Clearing the gate is worth 0.05 on its own, which is the only thing a policy can earn before it paints well enough to beat anything, and therefore the only reward variance a GRPO group has on day one. Failing the gate scores zero outright. A submission that loaded someone else's painting or wrote the answer in words does not collect the compile credit, and one that did not render has no painting to judge. Without a judge configured only the free components count, which is a thin signal useful for shaking out a run without spending on vision calls and not something to train against. """ if not self.gate.passed: return 0.0 total = GATE_WEIGHT + LENGTH_WEIGHT * self.length_score total += QUALITY_WEIGHT * self.quality_score if self.judge_enabled: total += JUDGE_WEIGHT * self.judge_score return total @property def critique(self) -> str: """`str`: What may be shown to the policy between revisions. Deliberately narrower than [`feedback`], which exists for a person reading a log and says how many references the painting beat and what it scored. Handing that to the policy would teach it to optimise against the specific references it was drawn, not to paint: with a shared seed the whole group faces the same eight, so naming the outcome names the opponents. The reference paintings and the score are the answer key. What is left is the submission's own properties, which it could have measured itself if it had eyes: whether it was admitted, why not, and how much of the canvas it covered. """ if not self.gate.passed: return "rejected: " + ", ".join(self.gate.violations) render = self.gate.render painted = render.paint_fraction if render else 0.0 return ( f"admitted. {painted:.1%} of the canvas carries pigment; the reference " f"paintings run from 12% to 25%." ) @property def feedback(self) -> str: """`str`: A short explanation of the score, safe to show the model.""" if not self.gate.passed: return "rejected: " + ", ".join(self.gate.violations) if not self.judge_enabled: return "painted, no judge configured" if not self.judged: return "painted, judge unavailable" # Counts only comparisons the judge actually resolved. Reporting a # loss for a reference nobody managed to look at reads as a verdict # when it is a failed call. resolved = [ c for c in self.judge.comparisons if c.score is not None and (c.submission_first is not None or c.reference_first is not None) ] won = sum(1 for c in resolved if c.score == 1.0) return ( f"painted, beat {won} of {len(resolved)} references" f" (judge {self.judge_score:.2f}, quality {self.quality_score:.2f}," f" reward {self.reward:.2f})" ) def to_dict(self) -> dict[str, Any]: """Return the full per-layer analysis.""" return { "task_id": self.task.task_id, "reward": self.reward, "components": { "gate": GATE_WEIGHT if self.gate.passed else 0.0, "length": LENGTH_WEIGHT * self.length_score, "quality": QUALITY_WEIGHT * self.quality_score, "judge": JUDGE_WEIGHT * self.judge_score if self.judge_enabled else 0.0, }, "gate": self.gate.to_dict(), "judge": self.judge.to_dict() if self.judge else None, "judge_enabled": self.judge_enabled, "judge_weight": JUDGE_WEIGHT if self.judge_enabled else 0.0, } async def evaluate_submission( response: str, task: Task, renderer: SketchRenderer, judge: PairwiseJudge | None, seed: int | None = None, references: int | None = None, scorer: QualityScorer | None = None, ) -> Evaluation: """Score one submission. Args: response (`str`): The model's raw reply. task ([`Task`]): The request it was answering. renderer ([`SketchRenderer`]): The browser to render with. judge ([`PairwiseJudge`], *optional*): The comparative scorer. `None` scores on gate admission alone. seed (`int`, *optional*): Makes the sampled references reproducible. references (`int`, *optional*): Override how many references to judge against. Defaults to the judge's own setting. scorer ([`QualityScorer`], *optional*): The absolute scorer. `None` leaves that term of the reward at zero. Returns: [`Evaluation`]: The assembled verdict. Examples: ```python evaluation = await evaluate_submission(reply, task, renderer, judge) print(evaluation.reward, evaluation.feedback) ``` """ gate = await run_gate(response, renderer) if not gate.passed: return Evaluation(task=task, gate=gate, judge_enabled=judge is not None) # Both vision terms look at the same painting, so they go out together rather # than one after the other. The absolute mark is one call against the pairwise # judge's two per reference, so it adds nothing to the wall clock. coroutines = [] if judge is not None: coroutines.append(judge.score(gate.png, seed=seed, references=references)) if scorer is not None: coroutines.append(scorer.score(gate.png)) results = list(await asyncio.gather(*coroutines)) if coroutines else [] report = results.pop(0) if judge is not None else None mark = results.pop(0) if scorer is not None else None return Evaluation( task=task, gate=gate, judge=report, judge_enabled=judge is not None, quality=mark, )