Scope boundary

Owns pre-release behavioral promotion validation of an exact RL candidate over complete policy-defined episodes in an isolated environment. It does not build the generic sandbox or perform production detection.

AID-H-012.005: AID-H-012.003 validates the candidate's complete behavior and side effects in representative episodes; AID-H-012.005 focuses on gaming of reward models, judges, and verifiers. Reward gaming is one failure class inside broader behavioral promotion validation.

AID-D-013: AID-H-012.003 blocks unsafe RL candidates before release; AID-D-013 detects reward and policy manipulation from observed reward streams and trajectories. Promotion validation and operational detection use different populations and outcomes.

Implementation guidance

• AID-H-012.003-G001: Run the complete policy-defined behavioral episode suite in an isolated environment and bind reward, action, side-effect, and constraint evidence to the exact candidate before promotion.

AID-H-012.003-G001

Apply this promotion control to a trainable sequential agent that can be executed in a representative, isolated simulator or sandbox and can emit per-episode behavioral telemetry. It observes how reward was earned; it does not replace reward design, runtime containment, or incident detection. A non-agentic model, an offline policy with no executable environment, or a workload with no meaningful episode semantics is NOT_APPLICABLE. Test-case mix, metrics, quantiles, and thresholds are defined by signed environment policy rather than framework-wide percentages.

Audit the exact sandbox episode population

# File: rl_audit/behavior_promotion_gate.py
from __future__ import annotations

import hashlib
import json
import math
import os
from pathlib import Path

from typing import Any

import polars as pl


class InsufficientData(RuntimeError):
    pass


def canonical_sha256(value: object) -> str:
    payload = json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode()
    return hashlib.sha256(payload).hexdigest()


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def audit_behavior(
    *,
    episodes_path: Path,
    population_manifest_path: Path,
    policy: dict[str, Any],
    verified_policy_sha256: str,
    sandbox_image_digest: str,
) -> dict:
    if canonical_sha256(policy) != verified_policy_sha256:
        raise ValueError("effective audit policy digest mismatch")
    if not sandbox_image_digest.startswith("sha256:") or len(sandbox_image_digest) != 71:
        raise ValueError("exact sandbox image digest is required")
    required_policy = {
        "policy_version",
        "top_reward_fraction",
        "max_top_reward_bad_rate",
        "max_top_reward_repetition_ratio",
        "minimum_scenario_success_rate",
    }
    if set(policy) != required_policy:
        raise ValueError("audit policy schema is invalid")
    numeric = [policy[name] for name in required_policy if name != "policy_version"]
    if not all(isinstance(value, (int, float)) and math.isfinite(value) for value in numeric):
        raise ValueError("audit thresholds must be finite")
    if not 0.0 < policy["top_reward_fraction"] <= 1.0:
        raise ValueError("top_reward_fraction must be in (0,1]")
    if any(not 0.0 <= policy[name] <= 1.0 for name in [
        "max_top_reward_bad_rate",
        "max_top_reward_repetition_ratio",
        "minimum_scenario_success_rate",
    ]):
        raise ValueError("rate thresholds must be in [0,1]")

    manifest = json.loads(population_manifest_path.read_text(encoding="utf-8"))
    expected_rows = manifest.get("episodes")
    if not isinstance(expected_rows, list) or not expected_rows:
        raise InsufficientData("population manifest has no expected episodes")
    expected = {
        (str(row["scenario_id"]), int(row["seed"]))
        for row in expected_rows
    }
    if len(expected) != len(expected_rows):
        raise ValueError("population manifest contains duplicate scenario/seed rows")

    episodes = pl.read_parquet(episodes_path)
    required_columns = {
        "episode_id",
        "scenario_id",
        "seed",
        "candidate_sha256",
        "total_reward",
        "intended_outcome_met",
        "safety_violation_count",
        "repetition_ratio",
        "step_count",
        "termination_reason",
        "telemetry_complete",
    }
    if set(episodes.columns) != required_columns:
        raise InsufficientData("episode schema or measured field population is incomplete")
    if episodes.height != episodes["episode_id"].n_unique():
        raise ValueError("episode IDs are duplicated")
    candidates = episodes["candidate_sha256"].unique().to_list()
    if len(candidates) != 1 or len(candidates[0]) != 64:
        raise ValueError("episodes must bind one exact candidate artifact")
    actual = set(
        zip(
            episodes["scenario_id"].cast(pl.String).to_list(),
            episodes["seed"].cast(pl.Int64).to_list(),
        )
    )
    if actual != expected or episodes.height != len(expected):
        raise InsufficientData("executed scenario/seed population differs from manifest")
    if not episodes["telemetry_complete"].cast(pl.Boolean).all():
        raise InsufficientData("one or more episodes has incomplete telemetry")
    if episodes.select(
        pl.any_horizontal(
            pl.col("total_reward").is_nan(),
            pl.col("total_reward").is_infinite(),
            pl.col("repetition_ratio").is_nan(),
            pl.col("repetition_ratio").is_infinite(),
        ).any()
    ).item():
        raise ValueError("behavior metrics contain non-finite values")
    if episodes.filter(
        (pl.col("repetition_ratio") < 0)
        | (pl.col("repetition_ratio") > 1)
        | (pl.col("safety_violation_count") < 0)
        | (pl.col("step_count") <= 0)
    ).height:
        raise ValueError("episode metrics are outside their valid domains")

    top_count = max(1, math.ceil(episodes.height * policy["top_reward_fraction"]))
    top = episodes.sort(
        ["total_reward", "episode_id"], descending=[True, False]
    ).head(top_count)
    top_bad = top.filter(
        (~pl.col("intended_outcome_met"))
        | (pl.col("safety_violation_count") > 0)
    ).height / top_count
    top_repetition = float(top["repetition_ratio"].max())

    by_scenario = (
        episodes.group_by("scenario_id")
        .agg(
            pl.len().alias("episode_count"),
            pl.col("intended_outcome_met").mean().alias("success_rate"),
            pl.col("safety_violation_count").sum().alias("violation_count"),
        )
        .sort("scenario_id")
    )
    findings: list[dict] = []
    if top_bad > policy["max_top_reward_bad_rate"]:
        findings.append({"check": "top_reward_bad_rate", "observed": top_bad})
    if top_repetition > policy["max_top_reward_repetition_ratio"]:
        findings.append({"check": "top_reward_repetition_ratio", "observed": top_repetition})
    for row in by_scenario.iter_rows(named=True):
        if row["success_rate"] < policy["minimum_scenario_success_rate"]:
            findings.append(
                {
                    "check": "scenario_success_rate",
                    "scenario_id": row["scenario_id"],
                    "observed": row["success_rate"],
                }
            )
        if row["violation_count"] > 0:
            findings.append(
                {
                    "check": "scenario_safety_violation",
                    "scenario_id": row["scenario_id"],
                    "observed": row["violation_count"],
                }
            )

    report = {
        "schema_version": "aidefend.rl-behavior-promotion.v1",
        "status": "FAIL" if findings else "PASS",
        "policy_version": policy["policy_version"],
        "policy_sha256": verified_policy_sha256,
        "candidate_sha256": candidates[0],
        "sandbox_image_digest": sandbox_image_digest,
        "population_manifest_sha256": file_sha256(population_manifest_path),
        "episodes_sha256": file_sha256(episodes_path),
        "expected_episode_count": len(expected),
        "observed_episode_count": episodes.height,
        "top_episode_count": top_count,
        "top_reward_bad_rate": top_bad,
        "top_reward_max_repetition_ratio": top_repetition,
        "scenario_summary": by_scenario.to_dicts(),
        "findings": findings,
    }
    return report

Verification, evidence, and outcomes

The sandbox runner must write immutable raw episode events first and derive the Parquet table from those events; retain both digests, the signed scenario/population and audit policies, signature-verification receipts, exact candidate and sandbox image, dependency lock, seeds, environment snapshot, videos or state traces required by policy, report, and promotion decision. A separate identity reloads the raw events, regenerates the table, recomputes the complete report, and verifies that the promoted artifact digest matches the candidate.

Action: Execute the exact candidate across the complete signed sandbox population, preserve raw behavior and derived evidence, and block promotion unless an independent regeneration and replay returns PASS.

Open-source tools

• Polars
• Gymnasium
• Ray RLlib
• Stable-Baselines3

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• Manipulation of Evaluation Metrics (L5)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.027 Misaligned Outputs

Cisco Integrated AI Security and Safety Framework

• AISubtech-2.2.2 Goal Substitution
• AISubtech-2.2.3 Authorization and Constraint Erosion
• AISubtech-2.3.1 Task Loophole Exploitation
• AITech-2.2 Goal Drift
• AITech-2.3 Reward Hacking

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (sandbox promotion audit blocks candidates with unintended actions)

Databricks AI Security Framework 3.0

• Agents - Core 13.7: Misaligned & Deceptive Behaviors