Scope boundary

Owns the explicit programmatic reward contract, including positive objectives, negative penalties, shaping terms, shortcut tests, versioning, and evidence that high reward corresponds to intended behavior.

AID-H-012.002: AID-H-012.001 engineers explicit environment or task reward logic; AID-H-012.002 governs a learned reward function derived from preferences or demonstrations. Programmatic reward design and learned signal calibration have different data and failure modes.

AID-H-012.005: AID-H-012.001 prevents shortcuts in the reward function itself; AID-H-012.005 prevents policies from gaming learned reward models, AI judges, or programmatic verifiers during modern post-training. A well-specified base reward does not eliminate evaluator-specific optimization exploits.

Implementation guidance

• AID-H-012.001-G001: Design the RL reward function as a single shaped objective that combines positive multi-objective goals with explicit penalties for unsafe or forbidden behaviors.
• AID-H-012.001-G002: Apply potential-based reward shaping only as a reviewed reward-engineering method with logged base rewards, potential deltas, and shortcut-farming tests.

AID-H-012.001-G001

Apply this control when a learned policy is optimized from an organization-defined scalar reward. The reward evaluator owns the exact shaped objective and training-episode termination signal; it does not replace runtime authorization, collision avoidance, safety interlocks, or AID-I-* containment. A non-RL workload or an RL method without an organization-controlled reward function is NOT_APPLICABLE. All weights, metric bounds, and red-line events come from a signature-verified, versioned environment policy rather than universal framework constants.

Evaluate one signed objective and emit a transition receipt

# File: rl_rewards/policy_bound_reward.py
from __future__ import annotations

import hashlib
import json
import math
from collections.abc import Mapping, Sequence
from typing import Any


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


def validate_policy(policy: Mapping[str, Any]) -> None:
    if not isinstance(policy.get("policy_version"), str) or not policy["policy_version"]:
        raise ValueError("policy_version is required")
    terms = policy.get("terms")
    red_lines = policy.get("red_lines")
    if not isinstance(terms, list) or not terms:
        raise ValueError("at least one objective or cost term is required")
    if not isinstance(red_lines, list) or not red_lines:
        raise ValueError("at least one explicit red-line event is required")

    metric_names: set[str] = set()
    for term in terms:
        if set(term) != {"metric", "weight", "minimum", "maximum"}:
            raise ValueError("reward term schema is invalid")
        name = term["metric"]
        if not isinstance(name, str) or not name or name in metric_names:
            raise ValueError("reward metric names must be non-empty and unique")
        values = [term["weight"], term["minimum"], term["maximum"]]
        if not all(not isinstance(value, bool) and isinstance(value, (int, float)) and math.isfinite(value) for value in values):
            raise ValueError("reward term values must be finite numbers")
        if float(term["minimum"]) > float(term["maximum"]):
            raise ValueError("reward metric bounds are reversed")
        metric_names.add(name)

    event_names: set[str] = set()
    for item in red_lines:
        if set(item) != {"event", "penalty", "terminate_episode"}:
            raise ValueError("red-line schema is invalid")
        name = item["event"]
        if not isinstance(name, str) or not name or name in event_names:
            raise ValueError("red-line event names must be non-empty and unique")
        if isinstance(item["penalty"], bool) or not isinstance(item["penalty"], (int, float)) or not math.isfinite(item["penalty"]):
            raise ValueError("red-line penalty must be finite")
        if float(item["penalty"]) >= 0 or item["terminate_episode"] is not True:
            raise ValueError("every red line requires a negative penalty and termination")
        event_names.add(name)


def evaluate_reward(
    *,
    metrics: Mapping[str, float],
    observed_events: Sequence[str],
    policy: Mapping[str, Any],
    verified_policy_sha256: str,
    episode_id: str,
    transition_id: str,
    candidate_policy_sha256: str,
) -> tuple[float, bool, dict]:
    validate_policy(policy)
    if canonical_sha256(policy) != verified_policy_sha256:
        raise ValueError("effective reward policy digest mismatch")
    if not episode_id or not transition_id:
        raise ValueError("episode_id and transition_id are required")
    if (not isinstance(candidate_policy_sha256, str) or
            __import__("re").fullmatch(r"[0-9a-f]{64}", candidate_policy_sha256) is None):
        raise ValueError("candidate policy digest must be lowercase SHA-256")

    expected_metrics = {term["metric"] for term in policy["terms"]}
    if set(metrics) != expected_metrics:
        raise ValueError("metrics must exactly match the signed reward-term population")
    red_line_names = {item["event"] for item in policy["red_lines"]}
    event_list = list(observed_events)
    if len(event_list) != len(set(event_list)):
        raise ValueError("observed_events contains duplicates")
    if not set(event_list).issubset(red_line_names):
        raise ValueError("an event is outside the signed event vocabulary")

    contributions: list[dict] = []
    total = 0.0
    for term in policy["terms"]:
        value = metrics[term["metric"]]
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
            raise ValueError(f"metric {term['metric']} is not finite")
        if not float(term["minimum"]) <= float(value) <= float(term["maximum"]):
            raise ValueError(f"metric {term['metric']} is outside its signed bounds")
        contribution = float(term["weight"]) * float(value)
        total += contribution
        contributions.append(
            {
                "kind": "objective_or_cost",
                "name": term["metric"],
                "observed": float(value),
                "contribution": contribution,
            }
        )

    terminate = False
    for item in policy["red_lines"]:
        triggered = item["event"] in event_list
        contribution = float(item["penalty"]) if triggered else 0.0
        total += contribution
        terminate = terminate or triggered
        contributions.append(
            {
                "kind": "red_line",
                "name": item["event"],
                "triggered": triggered,
                "contribution": contribution,
            }
        )
    if not math.isfinite(total):
        raise RuntimeError("computed reward is non-finite")

    receipt = {
        "schema_version": "aidefend.rl-shaped-reward-transition.v1",
        "status": "evaluated",
        "episode_id": episode_id,
        "transition_id": transition_id,
        "candidate_policy_sha256": candidate_policy_sha256,
        "reward_policy_version": policy["policy_version"],
        "reward_policy_sha256": verified_policy_sha256,
        "metrics_sha256": canonical_sha256(dict(metrics)),
        "observed_events": sorted(event_list),
        "contributions": contributions,
        "total_reward": total,
        "terminate_episode": terminate,
    }
    return total, terminate, receipt

Enforcement, verification, and evidence

The environment adapter must atomically persist the receipt and apply terminate_episode before accepting the transition into the training buffer. Retain the signed reward policy and verification receipt, evaluator image and dependency digests, candidate-policy digest, complete episode/transition manifest, all transition receipts, buffer-admission readback, and adversarial scenario results in append-only evidence storage. A separate verifier reloads the exact policy and transitions, recomputes every contribution and digest, confirms that all red-line transitions were terminated and excluded from subsequent training steps, and replays the complete signed normal/shortcut/red-line scenario population.

Action: Bind all positive terms, costs, and terminating red lines into one signed reward contract, enforce and record it on every transition, and block candidate promotion until an independent full-population replay returns PASS.

AID-H-012.001-G002

Potential-based reward shaping (PBRS) is applicable only when the organization controls a Markov reward process, can define a reviewed potential over the deployed state semantics, and can prove that the learner uses the same discount factor. It is a method within the signed reward contract in AID-H-012.001-G001, not a separate production safety barrier. An incompatible environment, unknown discount, or algorithm that does not consume additive per-transition rewards is NOT_APPLICABLE. Environment-specific bounds and evaluation thresholds come from signed policy.

Implement exact PBRS with terminal-state handling and receipts

# File: rl_rewards/pbrs_wrapper.py
from __future__ import annotations

import hashlib
import json
import math
from dataclasses import asdict, dataclass
from typing import Any

import gymnasium as gym
import numpy as np


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


@dataclass(frozen=True)
class PBRSConfig:
    policy_version: str
    gamma: float
    goal_state: tuple[float, ...]
    max_abs_shaping_reward: float
    truncation_is_terminal: bool
    replay_tolerance: float

    def validate(self) -> None:
        if not self.policy_version or not self.goal_state:
            raise ValueError("policy_version and goal_state are required")
        values = [*self.goal_state, self.gamma, self.max_abs_shaping_reward, self.replay_tolerance]
        if not all(math.isfinite(float(value)) for value in values):
            raise ValueError("PBRS policy values must be finite")
        if not 0.0 < self.gamma <= 1.0:
            raise ValueError("gamma must be in (0,1]")
        if self.max_abs_shaping_reward <= 0 or self.replay_tolerance <= 0:
            raise ValueError("bounds and replay tolerance must be positive")


def position(observation: dict[str, Any]) -> np.ndarray:
    value = np.asarray(observation["position"], dtype=np.float64)
    if value.ndim != 1 or not value.size or not np.isfinite(value).all():
        raise ValueError("observation position must be a finite vector")
    return value


def potential(observation: dict[str, Any], goal_state: tuple[float, ...]) -> float:
    current = position(observation)
    goal = np.asarray(goal_state, dtype=np.float64)
    if current.shape != goal.shape:
        raise ValueError("observation and signed goal dimensions differ")
    return -float(np.linalg.norm(current - goal))


def observation_sha256(observation: dict[str, Any]) -> str:
    return canonical_sha256({"position": position(observation).tolist()})


class PotentialBasedRewardShaping(gym.Wrapper):
    def __init__(
        self,
        env: gym.Env,
        config: PBRSConfig,
        verified_policy_sha256: str,
        learner_gamma: float,
    ):
        config.validate()
        if canonical_sha256(asdict(config)) != verified_policy_sha256:
            raise ValueError("effective PBRS policy digest mismatch")
        if not math.isclose(learner_gamma, config.gamma, rel_tol=0.0, abs_tol=config.replay_tolerance):
            raise ValueError("learner discount does not match PBRS gamma")
        super().__init__(env)
        self.config = config
        self.policy_sha256 = verified_policy_sha256
        self._last_potential: float | None = None
        self._last_observation_sha256: str | None = None
        self._episode_id: str | None = None
        self._step_number = 0

    def reset(self, *, episode_id: str, **kwargs):
        if not episode_id:
            raise ValueError("episode_id is required")
        observation, info = self.env.reset(**kwargs)
        self._last_potential = potential(observation, self.config.goal_state)
        self._last_observation_sha256 = observation_sha256(observation)
        self._episode_id = episode_id
        self._step_number = 0
        return observation, info

    def step(self, action):
        if self._last_potential is None or self._last_observation_sha256 is None:
            raise RuntimeError("PBRS wrapper used before reset")
        observation, base_reward, terminated, truncated, info = self.env.step(action)
        if isinstance(base_reward, bool) or not isinstance(base_reward, (int, float)) or not math.isfinite(base_reward):
            raise RuntimeError("base reward is non-finite")

        terminal_for_pbrs = bool(terminated or (truncated and self.config.truncation_is_terminal))
        next_potential = (
            0.0
            if terminal_for_pbrs
            else potential(observation, self.config.goal_state)
        )
        shaping_reward = (
            self.config.gamma * next_potential - self._last_potential
        )
        if not math.isfinite(shaping_reward):
            raise RuntimeError("PBRS shaping reward is non-finite")
        if abs(shaping_reward) > self.config.max_abs_shaping_reward:
            raise RuntimeError("PBRS shaping reward exceeds the signed bound")
        total_reward = float(base_reward) + shaping_reward
        next_sha256 = observation_sha256(observation)

        receipt = {
            "schema_version": "aidefend.pbrs-transition.v1",
            "status": "evaluated",
            "episode_id": self._episode_id,
            "step_number": self._step_number,
            "policy_version": self.config.policy_version,
            "policy_sha256": self.policy_sha256,
            "state_before_sha256": self._last_observation_sha256,
            "state_after_sha256": next_sha256,
            "potential_before": self._last_potential,
            "potential_after": next_potential,
            "base_reward": float(base_reward),
            "shaping_reward": shaping_reward,
            "total_reward": total_reward,
            "terminated": bool(terminated),
            "truncated": bool(truncated),
            "terminal_for_pbrs": terminal_for_pbrs,
        }
        self._last_potential = next_potential
        self._last_observation_sha256 = next_sha256
        self._step_number += 1
        output_info = dict(info)
        output_info["pbrs_receipt"] = receipt
        return observation, total_reward, terminated, truncated, output_info


def verify_episode(
    receipts: list[dict],
    *,
    expected_transition_count: int,
    config: PBRSConfig,
    verified_policy_sha256: str,
) -> dict:
    config.validate()
    if canonical_sha256(asdict(config)) != verified_policy_sha256:
        raise ValueError("verification policy digest mismatch")
    if len(receipts) != expected_transition_count or not receipts:
        raise ValueError("episode receipt population is incomplete")
    for index, receipt in enumerate(receipts):
        if receipt["step_number"] != index or receipt["policy_sha256"] != verified_policy_sha256:
            raise ValueError("receipt order or policy binding mismatch")
        expected = config.gamma * float(receipt["potential_after"]) - float(receipt["potential_before"])
        if not math.isclose(
            expected,
            float(receipt["shaping_reward"]),
            rel_tol=0.0,
            abs_tol=config.replay_tolerance,
        ):
            raise AssertionError("PBRS identity replay failed")
        if not math.isclose(
            float(receipt["base_reward"]) + expected,
            float(receipt["total_reward"]),
            rel_tol=0.0,
            abs_tol=config.replay_tolerance,
        ):
            raise AssertionError("total reward replay failed")
        if index and receipts[index - 1]["state_after_sha256"] != receipt["state_before_sha256"]:
            raise ValueError("transition chain is discontinuous")
    return {
        "schema_version": "aidefend.pbrs-episode-verification.v1",
        "status": "verified",
        "policy_sha256": verified_policy_sha256,
        "episode_id": receipts[0]["episode_id"],
        "transition_count": len(receipts),
        "receipts_sha256": canonical_sha256(receipts),
    }

Evidence, independent tests, and outcomes

Atomically store every receipt with the base-reward receipt, signed PBRS policy and signature-verification record, exact learner/configuration and environment image digests, seed, episode manifest, and candidate checkpoint. A separate runner checks learner gamma readback, verifies every PBRS identity and transition chain, and replays the signed complete normal, reversible-cycle, truncation, terminal, detour, and shortcut scenario population. Promotion also requires the policy-owned base-task utility, shaping-dependence, and shortcut-resistance thresholds; the framework supplies no universal percentage.

Action: Bind PBRS to the exact learner discount and environment semantics, preserve base and shaping rewards separately for every transition, and block promotion until independent full-population replay returns PASS.

Open-source tools

• Gymnasium
• NumPy
• Stable-Baselines3

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• N/A

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.3.1 Task Loophole Exploitation
• AITech-2.3 Reward Hacking

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions

Databricks AI Security Framework 3.0

• Agents - Core 13.7: Misaligned & Deceptive Behaviors