DEV Community

Finley Zhou
Finley Zhou

Posted on

Block Agent Merges That Shrink Properties or Silence Flakes

A green CI run is not a score for an agent patch. The patch can drop max_examples, wrap generators in assume(), or wrap a flake in time.sleep, and still exit 0. Freeze three numbers on a host the agent cannot write: property surface area, per-nodeid outcome entropy, and kill count on a pinned fault corpus. If any number moves in the wrong direction, reject the merge.

This is a proposed gate, not a field study. The scripts below are labeled examples. Run them against a parent tree and a patched tree before you treat the output as a merge signal.

Green hides two failure modes

Property tests do useful work in two ways. They emit a counterexample, or they stay loud when the implementation is wrong. Agent patches attack both without touching the test name.

Shrinkage is quiet. The generator still runs. The suite still “covers” the function. The input space is smaller, so the old bug no longer appears. Entropy laundering is louder. A flake that failed 3 of 11 times becomes a pass after a retry loop, a broader except, or a 50ms sleep. The process exit code cannot tell these apart from a real fix.

Test names are the wrong freeze key. Agents rename, split, and hollow functions while keeping the nodeid pattern that your dashboard already trusts. Hash the callable. Count the filters. Replay the nodeid. Do not score the label.

Decision table

Signal Freeze key Merge if it moves?
Property surface AST dump + assume count + max_examples No, if surface shrinks
Fixture bytes sha256 of declared paths No
Flake entropy 11 identical replays of a nodeid No; quarantine
Kill count pinned one-line faults Must not fall
New tautology assert True, empty @given, constant examples Reject

The agent may edit production code. It may not edit the files that produce these rows. Put those files in a path the patch reviewer treats as a protected surface, the same way you treat .github/workflows.

1. Measure shrinkage, not the test title

Walk every function decorated with @given. Record a tuple the agent does not get to choose: a normalized AST hash, the number of assume calls, and max_examples if a @settings decorator set it. Names are display only.

# score_surface.py — proposed gate, run on parent and patch trees
from __future__ import annotations

import ast
import hashlib
import json
import sys
from pathlib import Path


def _dump(node: ast.AST) -> str:
    return ast.dump(node, annotate_fields=True, include_attributes=False)


def _max_examples(deco: ast.AST) -> int | None:
    if not isinstance(deco, ast.Call):
        return None
    func = deco.func
    name = func.id if isinstance(func, ast.Name) else getattr(func, "attr", "")
    if name != "settings":
        return None
    for kw in deco.keywords:
        if kw.arg == "max_examples" and isinstance(kw.value, ast.Constant):
            if isinstance(kw.value.value, int):
                return kw.value.value
    return None


def _is_given(deco: ast.AST) -> bool:
    if not isinstance(deco, ast.Call):
        return False
    func = deco.func
    name = func.id if isinstance(func, ast.Name) else getattr(func, "attr", "")
    return name == "given"


def surfaces(root: Path) -> dict[str, dict]:
    out: dict[str, dict] = {}
    for path in root.rglob("test_*.py"):
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            if not any(_is_given(d) for d in node.decorator_list):
                continue
            assumes = sum(
                1
                for n in ast.walk(node)
                if isinstance(n, ast.Call)
                and (
                    (isinstance(n.func, ast.Name) and n.func.id == "assume")
                    or (isinstance(n.func, ast.Attribute) and n.func.attr == "assume")
                )
            )
            examples = None
            for deco in node.decorator_list:
                found = _max_examples(deco)
                if found is not None:
                    examples = found
            key = f"{path.as_posix()}::{node.name}"
            blob = _dump(node).encode("utf-8")
            out[key] = {
                "ast_sha256": hashlib.sha256(blob).hexdigest(),
                "assume_count": assumes,
                "max_examples": examples,
                "lineno": node.lineno,
            }
    return out


def shrinkage(parent: dict, patch: dict) -> list[dict]:
    findings = []
    for key, before in parent.items():
        after = patch.get(key)
        if after is None:
            findings.append({"id": key, "reason": "property_removed"})
            continue
        if after["assume_count"] > before["assume_count"]:
            findings.append(
                {
                    "id": key,
                    "reason": "assume_count_rose",
                    "from": before["assume_count"],
                    "to": after["assume_count"],
                }
            )
        b_ex, a_ex = before["max_examples"], after["max_examples"]
        if b_ex is not None and a_ex is not None and a_ex < b_ex:
            findings.append(
                {
                    "id": key,
                    "reason": "max_examples_fell",
                    "from": b_ex,
                    "to": a_ex,
                }
            )
        if after["ast_sha256"] != before["ast_sha256"] and after["assume_count"] >= before["assume_count"]:
            # AST changed; still a review signal even if counts did not fall.
            findings.append({"id": key, "reason": "property_ast_changed"})
    return findings


if __name__ == "__main__":
    parent, patch = Path(sys.argv[1]), Path(sys.argv[2])
    report = shrinkage(surfaces(parent), surfaces(patch))
    print(json.dumps(report, indent=2))
    sys.exit(1 if report else 0)
Enter fullscreen mode Exit fullscreen mode

A rising assume count is not a refactor. It is a filter. A falling max_examples is not a speedup if you still claim the same property. AST-only edits without count changes are a human review queue, not an automatic reject, unless you also lock counterexample hashes in a separate file the agent cannot touch.

Run it as two checkouts, not as a working tree the agent still holds:

git worktree add /tmp/parent HEAD
git worktree add /tmp/patch FETCH_HEAD
python score_surface.py /tmp/parent /tmp/patch
Enter fullscreen mode Exit fullscreen mode

2. Quarantine by entropy. Do not let the agent heal the flake

Pick a nodeid. Replay it 11 times with the same pytest seed, the same worker count, and a wiped cache. Map pass to 1 and fail to 0. Binary entropy of the pass rate is the freeze metric. Zero means deterministic. A mixed bitstring means the test is not a score.

# score_entropy.py — proposed gate
from __future__ import annotations

import math
import subprocess
import sys
from pathlib import Path


def bernoulli_entropy(p: float) -> float:
    if p <= 0.0 or p >= 1.0:
        return 0.0
    return -(p * math.log2(p) + (1.0 - p) * math.log2(1.0 - p))


def replay(nodeid: str, runs: int = 11, seed: int = 20260922) -> dict:
    bits: list[int] = []
    for i in range(runs):
        proc = subprocess.run(
            [
                sys.executable,
                "-m",
                "pytest",
                "-q",
                "--cache-clear",
                f"--randomly-seed={seed}",
                nodeid,
            ],
            check=False,
        )
        bits.append(1 if proc.returncode == 0 else 0)
    p = sum(bits) / len(bits)
    h = bernoulli_entropy(p)
    return {
        "nodeid": nodeid,
        "bits": bits,
        "pass_rate": p,
        "entropy_bits": h,
        "quarantine": 0.0 < p < 1.0,
    }


if __name__ == "__main__":
    report = replay(sys.argv[1])
    print(report)
    if report["quarantine"]:
        Path("flake_quarantine.txt").write_text(report["nodeid"] + "\n", encoding="utf-8")
        sys.exit(2)
Enter fullscreen mode Exit fullscreen mode

--randomly-seed assumes pytest-randomly is installed. If it is not, pin order another way: one worker, no xdist, and an explicit --seed your runner already documents. The point is identical replays. Mixed outcomes go into flake_quarantine.txt. The agent may not edit that file. The agent may not add retries, sleeps, or xfail marks to those nodeids.

A human owns the flake. Silence is not a fix.

3. Keep a tiny kill map the suite must still hit

Pass/fail on the happy path does not prove the assertions still mean anything. Keep a pinned list of one-line faults. Apply each fault to a copy of the tree. The parent must fail. The patch must still fail. If the patch deleted the assertion that used to kill fault F2, the kill count drops and the gate fails.

{
  "faults": [
    {
      "id": "F1",
      "file": "src/billing/tax.py",
      "replace": "rate = 0.0",
      "with": "rate = 1.0"
    },
    {
      "id": "F2",
      "file": "src/billing/tax.py",
      "replace": "return round(net, 2)",
      "with": "return net"
    },
    {
      "id": "F3",
      "file": "src/billing/tax.py",
      "replace": "if country == \"US\":",
      "with": "if country == \"XX\":"
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode
# proposed: apply one fault per worktree, run the frozen nodeids only
python - <<'PY'
import json, pathlib, subprocess, sys, tempfile, shutil
from pathlib import Path
spec = json.loads(Path("kill_map.json").read_text())
root = Path(".").resolve()
live = 0
for fault in spec["faults"]:
    tmp = Path(tempfile.mkdtemp(prefix="kill-"))
    shutil.copytree(root, tmp, dirs_exist_ok=True, ignore=shutil.ignore_patterns(".git", ".venv"))
    target = tmp / fault["file"]
    text = target.read_text(encoding="utf-8")
    if fault["replace"] not in text:
        print(fault["id"], "MISSING_SITE")
        sys.exit(3)
    target.write_text(text.replace(fault["replace"], fault["with"], 1), encoding="utf-8")
    proc = subprocess.run([sys.executable, "-m", "pytest", "-q", "tests/test_tax_properties.py"], cwd=tmp)
    killed = proc.returncode != 0
    live += int(killed)
    print(fault["id"], "KILLED" if killed else "SURVIVED")
print("kill_count", live, "of", len(spec["faults"]))
sys.exit(0 if live == len(spec["faults"]) else 1)
PY
Enter fullscreen mode Exit fullscreen mode

Keep the corpus small. Hand-written faults beat generated mutants for this gate. Generated mutants drift. Drift is another thing an agent can game.

4. Split the edit plane from the score plane

Do not compute these numbers on the laptop that applied the patch. Caches, extra environment variables, and leftover .pytest_cache leak into entropy. Check out parent and patch on a separate host. That host does not accept writes from the agent session.

Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode's free model access and free server option fit this split if you need a score host that is not the tree the agent can push to. Use the server as the only machine allowed to print the shrinkage JSON, the entropy bitstrings, and the kill count. Do not give that host an edit token.

A model does not belong in the scoring loop. Free model access is useful for one job only: turning the JSON reject payload into a readable PR comment. The model must not regenerate tests, must not suggest assume() filters, and must not propose sleeps. Feed it the gate output as read-only text. If it cannot see kill_map.json, it cannot rewrite it.

5. Wire the three exits into one check

set -euo pipefail
python score_surface.py /tmp/parent /tmp/patch
while read -r nodeid; do
  python score_entropy.py "$nodeid" || true
done < frozen_nodeids.txt
if [ -s flake_quarantine.txt ]; then
  echo "quarantine moved; human only" >&2
  # fail if the patch also touched those files
  git diff --name-only HEAD~1 | grep -Ff flake_quarantine.txt && exit 1 || true
fi
python kill_map.py
Enter fullscreen mode Exit fullscreen mode

Exit non-zero on shrinkage that reduces surface, on a kill-count drop, or on a patch that edits a quarantined nodeid. Entropy that only quarantines, without a touching diff, is a yellow card for humans. It is not a license for the agent to continue.

Limitations, and who should skip this

This gate assumes pytest, Hypothesis-style @given, and a suite that already has at least a few properties worth freezing. Unit tests that are a single example with a literal expected value will always look “unshrunk.” They also will not kill a useful fault corpus. Do not bolt this onto a 20-test hobby repo and call the result an agent evaluation.

Binary entropy over 11 runs is not a flake oracle for tests that talk to the network, the clock, or a shared database. Those tests need a sealed host and recorded fixtures first. This article does not claim a measured flake rate for any public project. The 11-run window and the 0-vs-mixed rule are parameters. Tune them on your own history, or do not ship the gate.

Kill maps go stale when the implementation is supposed to change behavior. If the patch’s job is to change tax rounding, fault F2 must be rewritten by a human before the agent runs. An outdated kill map rejects good work. An agent-editable kill map accepts hollow work. Only the first failure is acceptable.

Skip the model-written PR comment if your review culture already reads JSON. Skip the remote host if the agent cannot reach your CI cache anyway and you already wipe it. The method still needs the three numbers. Hosting is an isolation detail, not the test.

What to merge

Merge when the implementation changed, the property surface did not shrink, quarantined nodeids were left alone, and the pinned faults still die. That is a score. A green pyramid with fewer assume filters removed, more examples dropped, and one sleep added is a weaker suite with a better dashboard. Keep the dashboard. Freeze the power.

Top comments (0)