Arena: the environment repository, and redaction-pressure

An environment is an eval you take the gradient of, so it carries obligations a
benchmark does not: a held-out slice, a counterweighted reward, and a demonstrated
0.0 floor and 1.0 ceiling. README states the three as house rules.

redaction-pressure is Forge's redact stage as a task. The first reward design scored
survivors — secrets caught, decoys kept, prose kept — and an empty ruleset tied the
best real attempt at 0.500, because destroying nothing keeps everything. Counting
removals instead puts inaction at 0.000, shredding at 0.380, and an oracle at 1.000.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

environments/redaction_pressure/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "redaction-pressure"
+version = "0.1.0"
+description = "redaction-pressure — write redaction rules graded on records the author never saw."
+requires-python = ">=3.11"
+dependencies = ["verifiers", "regex"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["redaction_pressure"]

environments/redaction_pressure/redaction_pressure/__init__.py

@@ -0,0 +1,3 @@
+from redaction_pressure.taskset import RedactionTaskset
+
+__all__ = ["RedactionTaskset"]

environments/redaction_pressure/redaction_pressure/corpus.py

@@ -0,0 +1,161 @@
+"""The corpus generator.
+
+Every task is one *stream* of records: a visible slice the agent reads, and a held-out
+slice it never sees, drawn from the same generator with the same secret classes and
+different instances. Both slices are built from one seed, so a task is reproducible from
+its integer id alone and nothing has to ship as data.
+
+The held-out slice is the whole point. Rules written by reading examples will pass the
+examples; the only question worth rewarding is whether they survive instances the author
+never saw. A generator gives that for free, and a fixed dataset cannot give it at all.
+
+Each record carries `secrets` (spans that MUST be removed) and `decoys` (spans that look
+like secrets to a careless rule and MUST survive). The decoys are not garnish — without
+them "redact everything" scores perfectly, and the reward stops measuring anything.
+"""
+
+from __future__ import annotations
+
+import random
+import string
+from dataclasses import dataclass, field
+
+# Held together deliberately: every secret class is paired with a decoy class that shares
+# its surface shape. A rule keyed on shape alone scores on one and pays on the other.
+FIRST = ["Dana", "Priya", "Marcus", "Ines", "Tomas", "Rui", "Nadia", "Owen"]
+LAST = ["Okafor", "Lindqvist", "Baptiste", "Moreau", "Sato", "Ahmed", "Vance", "Duarte"]
+CITY = ["lisbon", "osaka", "tallinn", "bogota", "leeds", "dakar"]
+PUBLIC_HOSTS = ["docs.python.org", "status.cloud.example", "www.postgresql.org"]
+
+
+@dataclass(frozen=True)
+class Span:
+    """One region of a record, by class, with the exact text occupying it."""
+
+    kind: str
+    text: str
+
+
+@dataclass
+class Record:
+    """One document, plus the ground truth about what is in it."""
+
+    body: str
+    secrets: list[Span] = field(default_factory=list)
+    decoys: list[Span] = field(default_factory=list)
+
+
+def _token(rng: random.Random, n: int, alphabet: str = string.ascii_lowercase + string.digits) -> str:
+    return "".join(rng.choice(alphabet) for _ in range(n))
+
+
+# --- the seven secret classes, each with its shape-matched decoy ------------------------
+# Each builder returns (sentence, secret span, decoy span). The decoy is what a rule that
+# matched on shape rather than meaning would also destroy.
+
+
+def _api_key(rng: random.Random) -> tuple[str, Span, Span]:
+    secret = f"sk-live-{_token(rng, 24)}"
+    # Shape-identical and public: the test key in our own README.
+    decoy = f"sk-test-{'0' * 24}"
+    return (
+        f"Customer pasted their key {secret} into the ticket; our published sample is "
+        f"{decoy} and should stay in the docs snippet.",
+        Span("api_key", secret),
+        Span("public_test_key", decoy),
+    )
+
+
+def _email(rng: random.Random) -> tuple[str, Span, Span]:
+    first, last = rng.choice(FIRST), rng.choice(LAST)
+    secret = f"{first.lower()}.{last.lower()}@{_token(rng, 6, string.ascii_lowercase)}.com"
+    decoy = "support@lumbridgecorp.com"
+    return (
+        f"Reply-to was {secret}; the shared alias {decoy} was on cc and is public.",
+        Span("email", secret),
+        Span("public_alias", decoy),
+    )
+
+
+def _phone(rng: random.Random) -> tuple[str, Span, Span]:
+    secret = f"+1-{rng.randint(200, 989)}-{rng.randint(200, 999)}-{rng.randint(1000, 9999)}"
+    # Same digit shape, not a phone number: an order reference.
+    decoy = f"ORD-{rng.randint(200, 989)}-{rng.randint(200, 999)}-{rng.randint(1000, 9999)}"
+    return (
+        f"Callback number {secret} against order {decoy}.",
+        Span("phone", secret),
+        Span("order_ref", decoy),
+    )
+
+
+def _card(rng: random.Random) -> tuple[str, Span, Span]:
+    secret = "-".join(f"{rng.randint(1000, 9999)}" for _ in range(4))
+    # Sixteen digits in four groups that is a build id, not a card.
+    decoy = f"build {rng.randint(1000, 9999)}.{rng.randint(1000, 9999)}.{rng.randint(1000, 9999)}.{rng.randint(1000, 9999)}"
+    return (
+        f"They read the card {secret} aloud on the call, on {decoy}.",
+        Span("card", secret),
+        Span("build_id", decoy),
+    )
+
+
+def _host(rng: random.Random) -> tuple[str, Span, Span]:
+    secret = f"{_token(rng, 5, string.ascii_lowercase)}-db-{rng.randint(1, 9)}.internal.lumbridge"
+    decoy = rng.choice(PUBLIC_HOSTS)
+    return (
+        f"Trace pointed at {secret}; the public mirror {decoy} was fine.",
+        Span("internal_host", secret),
+        Span("public_host", decoy),
+    )
+
+
+def _person(rng: random.Random) -> tuple[str, Span, Span]:
+    secret = f"{rng.choice(FIRST)} {rng.choice(LAST)}"
+    # A person-shaped name that is a product, not a customer.
+    decoy = "Lumbridge Arena"
+    return (
+        f"Escalated by {secret} after the {decoy} demo.",
+        Span("person", secret),
+        Span("product_name", decoy),
+    )
+
+
+def _guid(rng: random.Random) -> tuple[str, Span, Span]:
+    secret = "-".join(_token(rng, n, string.hexdigits.lower()[:16]) for n in (8, 4, 4, 4, 12))
+    # A GUID that is deliberately public: the canary probe id, which must survive redaction
+    # or the contamination check downstream loses the thing it looks for.
+    decoy = "canary-0000-0000-0000-000000000000"
+    return (
+        f"Session {secret} carried the canary {decoy}.",
+        Span("session_guid", secret),
+        Span("canary_id", decoy),
+    )
+
+
+BUILDERS = [_api_key, _email, _phone, _card, _host, _person, _guid]
+
+
+def build_record(rng: random.Random) -> Record:
+    """One record: two or three sentence-pairs, each contributing a secret and its decoy."""
+    picks = rng.sample(BUILDERS, rng.randint(2, 3))
+    sentences, secrets, decoys = [], [], []
+    for build in picks:
+        sentence, secret, decoy = build(rng)
+        sentences.append(sentence)
+        secrets.append(secret)
+        decoys.append(decoy)
+    city = rng.choice(CITY)
+    body = f"[ticket/{city}/{rng.randint(1000, 9999)}] " + " ".join(sentences)
+    return Record(body=body, secrets=secrets, decoys=decoys)
+
+
+def build_slices(seed: int, visible: int, held_out: int) -> tuple[list[Record], list[Record]]:
+    """The two slices of one task, from one seed.
+
+    Drawn from a single stream rather than two, so the held-out records are the *next*
+    ones the generator would have produced — not a differently-seeded population that
+    could differ in ways the agent could not have anticipated.
+    """
+    rng = random.Random(seed)
+    records = [build_record(rng) for _ in range(visible + held_out)]
+    return records[:visible], records[visible:]

environments/redaction_pressure/redaction_pressure/scan.py

@@ -0,0 +1,136 @@
+"""Applying the agent's rules, and measuring what they did.
+
+Everything is counted as *removals*, and that choice is the whole reward.
+
+The obvious framing — secrets caught, decoys survived, prose kept — pays a model that
+returns no rules at all: it destroys nothing, so it keeps everything. Measured that way
+an empty ruleset and a ruleset that shreds the corpus scored identically, and neither
+told a trainer anything. Counting removals fixes it, because a model that removes nothing
+has no true positives and therefore no precision.
+
+  removed_secrets   true positives — a seeded secret gone from the held-out text.
+  residual_hits     the ones still there. Forge's rule applies to the gate: a residual
+                    hit is exit 1, not a deduction.
+  collateral_hits   decoys destroyed. Each is shape-matched to a secret class, so this is
+                    what a rule keyed on shape pays and one keyed on meaning does not.
+  innocent_lost     ordinary words destroyed. The floor that stops `\\S+ -> [X]`, which
+                    otherwise has perfect recall.
+
+The agent supplies the patterns, so they are hostile input: `regex` is used rather than
+`re` for its `timeout=`, since `re` cannot be interrupted and one nested quantifier would
+otherwise hang a rollout until the harness killed it.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import regex
+
+from redaction_pressure.corpus import Record
+
+MAX_RULES = 24
+MAX_PATTERN_CHARS = 200
+PATTERN_TIMEOUT_S = 0.25
+
+
+@dataclass(frozen=True)
+class Rule:
+    pattern: str
+    replacement: str
+
+
+@dataclass
+class Outcome:
+    """What one ruleset did to one slice."""
+
+    residual_hits: int
+    secrets_total: int
+    collateral_hits: int
+    decoys_total: int
+    innocent_lost: int
+    innocent_total: int
+    invalid_rules: list[str]
+
+    @property
+    def removed_secrets(self) -> int:
+        return self.secrets_total - self.residual_hits
+
+    @property
+    def recall(self) -> float:
+        return self.removed_secrets / max(self.secrets_total, 1)
+
+    @property
+    def precision(self) -> float:
+        """Of everything this ruleset destroyed, how much of it should have been.
+
+        Zero for an empty ruleset — it destroyed nothing, so none of what it destroyed
+        was a secret. That is the case the survivor-counting version got wrong.
+        """
+        destroyed = self.removed_secrets + self.collateral_hits + self.innocent_lost
+        return self.removed_secrets / destroyed if destroyed else 0.0
+
+    @property
+    def clean(self) -> bool:
+        """The gate: every secret gone AND no decoy taken with it — the ruleset you could
+        actually ship. `forge redact`'s exit 1, with the collateral clause it implies."""
+        return (
+            self.residual_hits == 0
+            and self.collateral_hits == 0
+            and self.secrets_total > 0
+        )
+
+
+def compile_rules(rules: list[Rule]) -> tuple[list[tuple[regex.Pattern, str]], list[str]]:
+    """Compile what compiles; report the rest rather than failing the rollout.
+
+    A rollout that dies on one bad pattern teaches nothing — the agent's score should
+    reflect that the rule did no work, which is what dropping it does.
+    """
+    compiled, invalid = [], []
+    for rule in rules[:MAX_RULES]:
+        if len(rule.pattern) > MAX_PATTERN_CHARS:
+            invalid.append(f"{rule.pattern[:40]}…: over {MAX_PATTERN_CHARS} characters")
+            continue
+        try:
+            compiled.append((regex.compile(rule.pattern), rule.replacement))
+        except regex.error as exc:
+            invalid.append(f"{rule.pattern[:40]}: {exc}")
+    return compiled, invalid
+
+
+def apply_rules(body: str, compiled: list[tuple[regex.Pattern, str]]) -> str:
+    """Rules in the order given, each over the whole text. A pattern that times out is
+    skipped for that record rather than aborting the measurement."""
+    text = body
+    for pattern, replacement in compiled:
+        try:
+            text = pattern.sub(replacement, text, timeout=PATTERN_TIMEOUT_S)
+        except (TimeoutError, regex.error):
+            continue
+    return text
+
+
+def _innocent_text(record: Record) -> str:
+    """The record with every secret and decoy cut out — the part no rule should touch."""
+    text = record.body
+    for span in [*record.secrets, *record.decoys]:
+        text = text.replace(span.text, " ")
+    return text
+
+
+def measure(records: list[Record], rules: list[Rule]) -> Outcome:
+    compiled, invalid = compile_rules(rules)
+    residual = collateral = secrets = decoys = lost = total = 0
+    for record in records:
+        redacted = apply_rules(record.body, compiled)
+        secrets += len(record.secrets)
+        decoys += len(record.decoys)
+        residual += sum(1 for s in record.secrets if s.text in redacted)
+        collateral += sum(1 for d in record.decoys if d.text not in redacted)
+        # Fidelity is counted over whole words rather than characters: a rule that eats
+        # every space would otherwise keep most of its characters and look harmless.
+        innocent = [w for w in _innocent_text(record).split() if w]
+        total += len(innocent)
+        lost += sum(1 for w in innocent if w not in redacted)
+    return Outcome(residual, secrets, collateral, decoys, lost, total, invalid)

environments/redaction_pressure/redaction_pressure/taskset.py

@@ -0,0 +1,143 @@
+"""redaction-pressure: write redaction rules that survive records you never read.
+
+The agent reads a handful of support tickets seeded with secrets, and returns a ruleset.
+The ruleset is then run against the *next* records the same generator would have produced.
+That gap is the environment: rules fitted to the visible examples score well on the
+visible examples and are worth nothing, and only the held-out slice can tell the two apart.
+
+Three rewards, all counted over what the ruleset *destroyed*:
+
+  recall     seeded secrets removed from the held-out text.
+  precision  of everything destroyed, the share that was a secret. An empty ruleset
+             destroys nothing and therefore scores zero here — which is the point. A
+             first pass scored survivors instead, and an empty ruleset tied the best
+             real attempt at 0.500 because it had kept every decoy and every word.
+  gate       binary: every secret gone and no decoy taken with it. `forge redact`'s
+             exit 1, which is a verdict and not a deduction.
+
+No judge model and no runtime: the reward is a scanner, so a rollout costs one completion
+and grades in milliseconds. Deterministic from the task seed.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import ClassVar
+
+from pydantic import Field
+
+import verifiers.v1 as vf
+
+from redaction_pressure.corpus import build_slices
+from redaction_pressure.scan import MAX_RULES, Rule, measure
+
+SYSTEM = f"""You are writing redaction rules for a support-ticket corpus.
+
+Return a JSON array of at most {MAX_RULES} rules in one ```json code block. Each rule is
+an object with "pattern" (a Python regular expression) and "replacement" (the text that
+replaces every match).
+
+Your rules will NOT be graded on the tickets below. They will be run against different
+tickets from the same source, containing the same kinds of secret and the same kinds of
+lookalike. Removing every secret matters most; destroying text that only resembles a
+secret, or shredding ordinary prose, costs you.
+
+Example shape:
+```json
+[{{"pattern": "\\\\bfoo-\\\\d+\\\\b", "replacement": "[REF]"}}]
+```"""
+
+_BLOCK = re.compile(r"```(?:json)?\s*\n(.*?)```", re.DOTALL)
+
+
+def parse_rules(reply: str) -> list[Rule]:
+    """The last JSON array in the reply. A reply with no parsable ruleset is an empty
+    ruleset, not an error — it scores what doing nothing scores."""
+    candidates = _BLOCK.findall(reply or "")
+    raw = candidates[-1] if candidates else (reply or "")
+    try:
+        parsed = json.loads(raw.strip())
+    except json.JSONDecodeError:
+        return []
+    if not isinstance(parsed, list):
+        return []
+    rules = []
+    for item in parsed:
+        if isinstance(item, dict) and isinstance(item.get("pattern"), str):
+            rules.append(Rule(item["pattern"], str(item.get("replacement", ""))))
+    return rules
+
+
+class RedactionData(vf.TaskData):
+    seed: int
+    """Rebuilds both slices exactly; the held-out records are never serialized here."""
+    visible: int
+    held_out: int
+
+
+class RedactionTask(vf.Task[RedactionData]):
+    @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]:
+        """Run the ruleset over the held-out slice once; every reward reads this."""
+        _, held_out = build_slices(self.data.seed, self.data.visible, self.data.held_out)
+        outcome = measure(held_out, parse_rules(trace.last_reply))
+        return {
+            "recall": outcome.recall,
+            "precision": outcome.precision,
+            "clean": float(outcome.clean),
+            "residual_hits": float(outcome.residual_hits),
+            "collateral_hits": float(outcome.collateral_hits),
+            "innocent_lost": float(outcome.innocent_lost),
+            "invalid_rules": float(len(outcome.invalid_rules)),
+        }
+
+    @vf.reward(weight=0.35)
+    async def recall(self, trace: vf.Trace) -> float:
+        return trace.metrics.get("recall", 0.0)
+
+    @vf.reward(weight=0.35)
+    async def precision(self, trace: vf.Trace) -> float:
+        return trace.metrics.get("precision", 0.0)
+
+    @vf.reward(weight=0.30)
+    async def gate(self, trace: vf.Trace) -> float:
+        return trace.metrics.get("clean", 0.0)
+
+
+class RedactionConfig(vf.TasksetConfig):
+    num_tasks: int = Field(64, ge=1)
+    visible: int = Field(4, ge=1)
+    """Records the agent reads."""
+    held_out: int = Field(12, ge=1)
+    """Records it is graded on and never sees."""
+
+
+class RedactionTaskset(vf.Taskset[RedactionTask, RedactionConfig]):
+    SEED_BASE: ClassVar[int] = 90_000
+
+    def load(self) -> list[RedactionTask]:
+        tasks = []
+        for i in range(self.config.num_tasks):
+            seed = self.SEED_BASE + i
+            visible, _ = build_slices(seed, self.config.visible, self.config.held_out)
+            tickets = "\n\n".join(record.body for record in visible)
+            tasks.append(
+                RedactionTask(
+                    RedactionData(
+                        idx=i,
+                        name=f"stream-{seed}",
+                        prompt=f"Tickets from the stream:\n\n{tickets}\n\nWrite the ruleset.",
+                        system_prompt=SYSTEM,
+                        seed=seed,
+                        visible=self.config.visible,
+                        held_out=self.config.held_out,
+                    ),
+                    self.config.task,
+                )
+            )
+        return tasks