Before You Let a Managed Agent Ship: Designing Your Own Acceptance Gate

# acceptance_gate.py — inspect agent artifacts (keep the primary gate deterministic)
from __future__ import annotations
import json, re, sys, hashlib
from dataclasses import dataclass, field
from pathlib import Path
 
REQUIRED_FIELDS = ("title", "slug", "body", "tags")
MIN_BODY_CHARS = 1200          # sanity floor for thin artifacts
MIN_SIGNALS = 3                # minimum practical-utility signals
 
@dataclass
class GateResult:
    accepted: bool
    reasons: list[str] = field(default_factory=list)
 
    def reject(self, msg: str) -> None:
        self.accepted = False
        self.reasons.append(msg)
 
def _paragraphs(text: str) -> list[str]:
    return [p.strip() for p in re.split(r"\n\s*\n", text) if len(p.strip()) >= 80]
 
def _practical_signals(body: str) -> int:
    """Count things that give the reader real value: code, numbers, steps, recommendations."""
    signals = 0
    if (chr(96) * 3) in body:                       signals += 1   # runnable code (code fence)
    if re.search(r"\d+\s?(ms|MB|%|req/|tok)", body): signals += 1   # concrete measurements
    if re.search(r"^\s*\d+\.\s", body, re.M):        signals += 1   # numbered procedure
    if re.search(r"(I |in production|actually|first-hand)", body): signals += 1  # first-hand context
    return signals
 
def check_artifact(path: Path, corpus_hashes: set[str]) -> GateResult:
    res = GateResult(accepted=True)
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as e:
        res.reject(f"Cannot parse as JSON: {e}")
        return res
 
    # 1) Schema: required fields present
    for key in REQUIRED_FIELDS:
        if not data.get(key):
            res.reject(f"Required field '{key}' is empty or missing")
 
    body = data.get("body", "")
    # 2) Sanity: reject artifacts that are too thin
    if len(body) < MIN_BODY_CHARS:
        res.reject(f"Body is {len(body)} chars, below the {MIN_BODY_CHARS} floor")
 
    # 3) Practical signals
    n = _practical_signals(body)
    if n < MIN_SIGNALS:
        res.reject(f"Only {n} practical signals (need at least {MIN_SIGNALS})")
 
    # 4) Verbatim duplication against the existing corpus
    for para in _paragraphs(body):
        h = hashlib.sha256(para.encode("utf-8")).hexdigest()
        if h in corpus_hashes:
            res.reject("Contains a paragraph that matches a previous artifact verbatim")
            break
    return res
 
def load_corpus_hashes(accepted_dir: Path) -> set[str]:
    hashes: set[str] = set()
    for f in accepted_dir.glob("*.json"):
        body = json.loads(f.read_text(encoding="utf-8")).get("body", "")
        for para in _paragraphs(body):
            hashes.add(hashlib.sha256(para.encode("utf-8")).hexdigest())
    return hashes
 
if __name__ == "__main__":
    artifact = Path(sys.argv[1])
    accepted = Path(sys.argv[2]) if len(sys.argv) > 2 else Path("accepted")
    accepted.mkdir(exist_ok=True)
    result = check_artifact(artifact, load_corpus_hashes(accepted))
    if result.accepted:
        print("OK accepted")
        sys.exit(0)
    print("rejected")
    for r in result.reasons:
        print("  -", r)
    sys.exit(1)

# run_and_gate.py — run agent -> quarantine -> gate -> accepted
import shutil
from pathlib import Path
from acceptance_gate import check_artifact, load_corpus_hashes
 
QUARANTINE = Path("quarantine")
ACCEPTED = Path("accepted")
 
def run_managed_agent(task: str) -> Path:
    """
    Start a Managed Agent and pull its artifact into quarantine.
    The public-preview API may change, so keep ONLY this function swappable.
    In production, box the agents.create -> run -> list_files -> download
    flow inside here and always return a Path under quarantine.
    """
    QUARANTINE.mkdir(exist_ok=True)
    out = QUARANTINE / "artifact.json"
    # client.agents.run(model="antigravity-preview-05-2026", task=task, ...)
    # downloaded = client.agents.download(run_id, "out/article.json")
    # out.write_bytes(downloaded)
    return out
 
def accept_or_retry(task: str, max_attempts: int = 3) -> bool:
    ACCEPTED.mkdir(exist_ok=True)
    feedback = ""
    for attempt in range(1, max_attempts + 1):
        prompt = task if not feedback else f"{task}\n\nYour last output was rejected. Fix all of these:\n{feedback}"
        artifact = run_managed_agent(prompt)
        result = check_artifact(artifact, load_corpus_hashes(ACCEPTED))
        if result.accepted:
            shutil.move(str(artifact), ACCEPTED / artifact.name)
            print(f"accepted on attempt {attempt}")
            return True
        feedback = "\n".join(f"- {r}" for r in result.reasons)
        print(f"rejected on attempt {attempt}:\n{feedback}")
    print("attempt limit reached; routing to human review")
    return False

# secondary gate (optional): call only AFTER the primary gate passes
def llm_review(body: str, client) -> tuple[bool, str]:
    rubric = (
        "Inspect the following body. Flag template-y intros, repeated claims, "
        "and unsupported assertions. If there are none, reply with just OK."
    )
    resp = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=[rubric, body],
    )
    verdict = resp.text.strip()
    return verdict.startswith("OK"), verdict