Three more environments, and probe.py to gate all four
canary-trap, fault-localisation and schema-migration, in the three domains the Environments Hub has nothing in: contamination, operations, data engineering. None is a port. Each is graded on a slice the model never read. canary-trap runs a probe set against a model that memorised the corpus AND one that learned the same facts from a reworded copy with every identifier regenerated. The obvious probe -- plant a GUID, ask for it back -- is sound and caps at 0.525 because it cannot see the paraphrase. It scored higher than that at first: facts with no distinct rewording had the paraphrase restating the corpus verbatim, so GUID probes caught it by accident. Every fact now carries a "core" token present in both wordings and in no public one, asserted at import, which is also what makes the ceiling reachable. fault-localisation puts the fault upstream and the noise downstream, the way incidents actually present. The broken service logs one line and goes quiet while its dependents log a dozen timeouts, so ranking by error volume answers the victim -- 0/64 incidents have the loudest service as the root, and probe.py asserts it rather than trusting the generator to stay that way. schema-migration executes the model's SQL against forty held-out rows carrying thousands separators, negatives, a unit with a slash in it and a value with no unit. Row preservation started as a reward beside fidelity and paid 0.15 for adding two empty columns and touching nothing; it is a multiplier now, so losing rows still zeroes a perfect migration and keeping them earns nothing on its own. probe.py is house rule 3 as an exit code: 0.000 for inaction and 1.000 for an oracle, or it returns 1. It loads the leaf modules through a namespace shim rather than the packages, so it runs without verifiers installed -- a self-check that needs the training stack is a self-check nobody runs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,3 @@
|
||||
from canary_trap.taskset import CanaryTaskset
|
||||
|
||||
__all__ = ["CanaryTaskset"]
|
||||
@@ -0,0 +1,121 @@
|
||||
"""The three knowledge states a probe has to tell apart.
|
||||
|
||||
A contamination probe is only useful if it separates a model that has seen your eval set
|
||||
from one that has not. The trap is that the obvious probe — carry a GUID into the corpus
|
||||
and ask for it back — separates only the *verbatim* case. Training data gets reworded,
|
||||
deduplicated and paraphrased on the way into a model, and a model that learned every fact
|
||||
in your eval set from a reworded copy answers no GUID and is contaminated anyway.
|
||||
|
||||
So a task builds three knowledge states from one seed:
|
||||
|
||||
clean public facts only. Answering from here is not evidence of anything.
|
||||
verbatim public + the corpus exactly as written, GUIDs and all.
|
||||
paraphrase public + the same FACTS, reworded, with every GUID regenerated.
|
||||
|
||||
A probe set is scored on catching both contaminated states without firing on the clean
|
||||
one. Verbatim-only probes cap at half the detection score, and that ceiling is the whole
|
||||
lesson of the environment.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
import string
|
||||
from dataclasses import dataclass
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Fact:
|
||||
"""One fact in three parts.
|
||||
|
||||
`canonical` is how the corpus states it. `alias` is how the paraphrase states it, and
|
||||
it shares no wording with `canonical` beyond `core`. `core` is the token a probe has to
|
||||
key on to see BOTH — and it is deliberately absent from public knowledge, so keying on
|
||||
it is also sound. Every fact therefore has an answer that scores 1.0, and none of them
|
||||
is the sentence the corpus actually contains.
|
||||
"""
|
||||
|
||||
subject: str
|
||||
attribute: str
|
||||
canonical: str
|
||||
alias: str
|
||||
core: str
|
||||
|
||||
|
||||
SUBJECTS = [
|
||||
Fact("the Lisbon relay", "throughput", "412 requests per second", "412 rps", "412"),
|
||||
Fact("the Osaka mirror", "operator", "Ines Baptiste", "I. Baptiste", "Baptiste"),
|
||||
Fact("the Tallinn cache", "eviction policy", "least-recently-used (LRU)", "LRU ordering", "LRU"),
|
||||
Fact("the Bogota gateway", "timeout", "1400 milliseconds", "1400 ms", "1400"),
|
||||
Fact("the Leeds shard", "replica count", "seven replicas", "seven copies", "seven"),
|
||||
Fact("the Dakar queue", "retry budget", "nine retries", "nine attempts", "nine"),
|
||||
]
|
||||
|
||||
# Facts a model could know without ever seeing our corpus. A probe whose answer lives here
|
||||
# fires on a clean model, which is what makes it a false positive rather than a probe.
|
||||
PUBLIC = [
|
||||
"TCP uses a three-way handshake.",
|
||||
"Postgres defaults to port 5432.",
|
||||
"HTTP 429 means too many requests.",
|
||||
"UTC does not observe daylight saving.",
|
||||
"SHA-256 emits 32 bytes.",
|
||||
"The capital of Portugal is Lisbon.",
|
||||
]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Knowledge:
|
||||
"""What one model can answer from. Membership is a substring test on `text`."""
|
||||
|
||||
name: str
|
||||
text: str
|
||||
|
||||
|
||||
def _check_facts() -> None:
|
||||
"""The environment's own floor-and-ceiling guarantee, asserted at import.
|
||||
|
||||
Every fact must have an answer that a probe can key on to see both contaminated states
|
||||
(`core` in canonical AND in alias) without accusing a clean model (`core` absent from
|
||||
public). Without this the ceiling is unreachable for some seeds and the reward quietly
|
||||
stops being solvable — which is exactly the failure the house rules exist to catch.
|
||||
"""
|
||||
public = " ".join(PUBLIC).lower()
|
||||
for fact in SUBJECTS:
|
||||
assert fact.core.lower() in fact.canonical.lower(), fact
|
||||
assert fact.core.lower() in fact.alias.lower(), fact
|
||||
assert fact.core.lower() not in public, f"{fact.core} collides with public knowledge"
|
||||
# And the paraphrase must not simply restate the corpus, or a verbatim probe would
|
||||
# catch it for free and the environment would teach nothing.
|
||||
assert fact.alias.lower() != fact.canonical.lower(), fact
|
||||
|
||||
|
||||
_check_facts()
|
||||
|
||||
|
||||
def _guid(rng: random.Random) -> str:
|
||||
hexd = "0123456789abcdef"
|
||||
return "-".join("".join(rng.choice(hexd) for _ in range(n)) for n in (8, 4, 4, 12))
|
||||
|
||||
|
||||
def build(seed: int, facts: int) -> tuple[str, dict[str, Knowledge]]:
|
||||
"""One corpus, and the three knowledge states built from it."""
|
||||
rng = random.Random(seed)
|
||||
picked = rng.sample(SUBJECTS, min(facts, len(SUBJECTS)))
|
||||
public_text = " ".join(PUBLIC)
|
||||
|
||||
verbatim_lines, paraphrase_lines = [], []
|
||||
for fact in picked:
|
||||
verbatim_lines.append(
|
||||
f"[{_guid(rng)}] Record for {fact.subject}: {fact.attribute} is {fact.canonical}."
|
||||
)
|
||||
# Same fact, different sentence, and a REGENERATED tag — the paraphrase remembers
|
||||
# what is true and has forgotten every string that carried it.
|
||||
paraphrase_lines.append(
|
||||
f"[{_guid(rng)}] Concerning {fact.subject}, its {fact.attribute} sits at {fact.alias}."
|
||||
)
|
||||
|
||||
corpus = "\n".join(verbatim_lines)
|
||||
return corpus, {
|
||||
"clean": Knowledge("clean", public_text),
|
||||
"verbatim": Knowledge("verbatim", public_text + "\n" + corpus),
|
||||
"paraphrase": Knowledge("paraphrase", public_text + "\n" + "\n".join(paraphrase_lines)),
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Scoring a probe set against the three knowledge states.
|
||||
|
||||
Two properties, and they pull against each other:
|
||||
|
||||
sound the probe does NOT answer on the clean model. An unsound probe accuses a model
|
||||
that has seen nothing, which is worse than no probe at all — Bench VOIDS a
|
||||
score on a contamination verdict, so a false positive destroys a real result.
|
||||
catching a sound probe answers on a contaminated model. Both contaminated states have
|
||||
to be caught, and the paraphrase is the one that costs: it kept every fact and
|
||||
not one of the strings, so a GUID-recall probe cannot see it.
|
||||
|
||||
Detection is scored per state rather than per probe, because catching the same state with
|
||||
six probes is not six times the evidence. Specificity is per probe, so shotgunning a
|
||||
hundred guesses to land one is paid for.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from canary_trap.corpus import Knowledge
|
||||
|
||||
MAX_PROBES = 12
|
||||
MIN_ANSWER_CHARS = 3
|
||||
CONTAMINATED = ("verbatim", "paraphrase")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Probe:
|
||||
question: str
|
||||
answer: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Outcome:
|
||||
sound: int = 0
|
||||
total: int = 0
|
||||
caught: set[str] = field(default_factory=set)
|
||||
rejected: list[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def specificity(self) -> float:
|
||||
"""Share of probes that do not accuse a clean model. Zero probes scores zero: an
|
||||
empty set has no false positives and also no evidence, and rewarding it would make
|
||||
returning nothing the safe play."""
|
||||
return self.sound / self.total if self.total else 0.0
|
||||
|
||||
@property
|
||||
def detection(self) -> float:
|
||||
return len(self.caught) / len(CONTAMINATED)
|
||||
|
||||
@property
|
||||
def clean_gate(self) -> bool:
|
||||
"""Every contaminated state caught, and not one probe firing on the clean model."""
|
||||
return self.detection == 1.0 and self.total > 0 and self.sound == self.total
|
||||
|
||||
|
||||
def _answers(knowledge: Knowledge, probe: Probe) -> bool:
|
||||
"""The model stands in for itself: it can produce an answer exactly when that answer is
|
||||
present in what it knows. Crude, and the crudeness is load-bearing — it makes the whole
|
||||
environment deterministic and free, and it still distinguishes recall of a string from
|
||||
recall of a fact, which is the only distinction being taught."""
|
||||
return probe.answer.lower() in knowledge.text.lower()
|
||||
|
||||
|
||||
def measure(states: dict[str, Knowledge], probes: list[Probe]) -> Outcome:
|
||||
outcome = Outcome()
|
||||
for probe in probes[:MAX_PROBES]:
|
||||
# A blank or near-blank answer is a substring of everything, which would read as a
|
||||
# probe that catches all three states. Rejected rather than scored as unsound, so
|
||||
# the reason shows up in the trace.
|
||||
if len(probe.answer.strip()) < MIN_ANSWER_CHARS:
|
||||
outcome.rejected.append(probe.answer[:24])
|
||||
outcome.total += 1
|
||||
continue
|
||||
outcome.total += 1
|
||||
if _answers(states["clean"], probe):
|
||||
continue # unsound: fires on a model that has seen nothing
|
||||
outcome.sound += 1
|
||||
for name in CONTAMINATED:
|
||||
if _answers(states[name], probe):
|
||||
outcome.caught.add(name)
|
||||
return outcome
|
||||
|
||||
|
||||
_BLOCK = re.compile(r"```(?:json)?\s*\n(.*?)```", re.DOTALL)
|
||||
|
||||
|
||||
def parse_probes(reply: str) -> list[Probe]:
|
||||
blocks = _BLOCK.findall(reply or "")
|
||||
raw = blocks[-1] if blocks else (reply or "")
|
||||
try:
|
||||
parsed = json.loads(raw.strip())
|
||||
except json.JSONDecodeError:
|
||||
return []
|
||||
if not isinstance(parsed, list):
|
||||
return []
|
||||
return [
|
||||
Probe(str(p.get("question", "")), p["answer"])
|
||||
for p in parsed
|
||||
if isinstance(p, dict) and isinstance(p.get("answer"), str)
|
||||
]
|
||||
@@ -0,0 +1,110 @@
|
||||
"""canary-trap: write contamination probes that survive a paraphrase.
|
||||
|
||||
The agent reads a corpus destined to become an eval set and returns probes that would
|
||||
reveal a model trained on it. Probes are then run against three stand-in models: one that
|
||||
has seen only public facts, one that memorised the corpus verbatim, and one that learned
|
||||
the same facts from a reworded copy with every identifier regenerated.
|
||||
|
||||
The obvious probe — ask for a GUID that was planted in the corpus — is sound and catches
|
||||
exactly one of the two contaminated models, so it ceilings at 0.5 detection. The probe that
|
||||
scores is one keyed on a FACT the paraphrase kept, whose answer does not also appear in
|
||||
public knowledge. Learning that difference is the environment.
|
||||
|
||||
This is Bench's contamination gate as a task. There, a canary hit does not annotate a score,
|
||||
it voids it — which is why a probe that fires on a clean model is the expensive failure here
|
||||
and specificity carries as much weight as detection.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
import verifiers.v1 as vf
|
||||
|
||||
from canary_trap.corpus import build
|
||||
from canary_trap.scan import MAX_PROBES, Probe, measure, parse_probes
|
||||
|
||||
SYSTEM = f"""You are designing contamination probes for an evaluation corpus.
|
||||
|
||||
Return a JSON array of at most {MAX_PROBES} probes in one ```json code block. Each probe is
|
||||
an object with "question" (what you would ask a model) and "answer" (the exact string a
|
||||
model that has seen this corpus would produce).
|
||||
|
||||
Your probes are run against three models: one that has seen only public knowledge, one that
|
||||
memorised this corpus word for word, and one that learned the same facts from a reworded
|
||||
copy in which every bracketed identifier was regenerated.
|
||||
|
||||
A probe that the public-knowledge model can answer is worse than no probe — it accuses an
|
||||
innocent model. A probe only the word-for-word model can answer catches half of what is
|
||||
there.
|
||||
|
||||
Example shape:
|
||||
```json
|
||||
[{{"question": "What is the retry budget for the Dakar queue?", "answer": "three retries"}}]
|
||||
```"""
|
||||
|
||||
class CanaryData(vf.TaskData):
|
||||
seed: int
|
||||
facts: int
|
||||
|
||||
|
||||
class CanaryTask(vf.Task[CanaryData]):
|
||||
@vf.stop
|
||||
async def single_turn(self, trace: vf.Trace) -> bool:
|
||||
return trace.num_turns >= 1
|
||||
|
||||
@vf.metric
|
||||
async def scan(self, trace: vf.Trace) -> dict[str, float]:
|
||||
_, states = build(self.data.seed, self.data.facts)
|
||||
outcome = measure(states, parse_probes(trace.last_reply))
|
||||
return {
|
||||
"detection": outcome.detection,
|
||||
"specificity": outcome.specificity,
|
||||
"clean": float(outcome.clean_gate),
|
||||
"probes": float(outcome.total),
|
||||
"rejected": float(len(outcome.rejected)),
|
||||
# Reported separately because it is the finding the environment exists to
|
||||
# teach: a probe set can be perfectly sound and still blind to a paraphrase.
|
||||
"caught_paraphrase": float("paraphrase" in outcome.caught),
|
||||
}
|
||||
|
||||
@vf.reward(weight=0.35)
|
||||
async def detection(self, trace: vf.Trace) -> float:
|
||||
return trace.metrics.get("detection", 0.0)
|
||||
|
||||
@vf.reward(weight=0.35)
|
||||
async def specificity(self, trace: vf.Trace) -> float:
|
||||
return trace.metrics.get("specificity", 0.0)
|
||||
|
||||
@vf.reward(weight=0.30)
|
||||
async def gate(self, trace: vf.Trace) -> float:
|
||||
return trace.metrics.get("clean", 0.0)
|
||||
|
||||
|
||||
class CanaryConfig(vf.TasksetConfig):
|
||||
num_tasks: int = Field(48, ge=1)
|
||||
facts: int = Field(4, ge=1, le=6)
|
||||
|
||||
|
||||
class CanaryTaskset(vf.Taskset[CanaryTask, CanaryConfig]):
|
||||
SEED_BASE = 70_000
|
||||
|
||||
def load(self) -> list[CanaryTask]:
|
||||
tasks = []
|
||||
for i in range(self.config.num_tasks):
|
||||
seed = self.SEED_BASE + i
|
||||
corpus, _ = build(seed, self.config.facts)
|
||||
tasks.append(
|
||||
CanaryTask(
|
||||
CanaryData(
|
||||
idx=i,
|
||||
name=f"corpus-{seed}",
|
||||
prompt=f"The corpus:\n\n{corpus}\n\nWrite the probes.",
|
||||
system_prompt=SYSTEM,
|
||||
seed=seed,
|
||||
facts=self.config.facts,
|
||||
),
|
||||
self.config.task,
|
||||
)
|
||||
)
|
||||
return tasks
|
||||
@@ -0,0 +1,13 @@
|
||||
[project]
|
||||
name = "canary-trap"
|
||||
version = "0.1.0"
|
||||
description = "canary-trap — write contamination probes that survive a paraphrase, without accusing a clean model."
|
||||
requires-python = ">=3.11"
|
||||
dependencies = ["verifiers"]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["canary_trap"]
|
||||
Reference in New Issue
Block a user