Python API Reference

Top-level exports

import humanproof

humanproof

humanproof - Motor-noise fingerprinting + HITL approval gates for agents.

CalibratedMotorScorer(noise_threshold, correction_threshold)

Bases: MotorScorer

A MotorScorer with calibrated thresholds from labeled examples.

Source code in src/humanproof/calibration.py

def __init__(
    self,
    noise_threshold: float,
    correction_threshold: float,
) -> None:
    super().__init__()
    self._noise_threshold = noise_threshold
    self._correction_threshold = correction_threshold

ApprovalError

Bases: ValueError

Raised when an action is refused for missing/invalid approval.

ApprovalSession(*, high_risk_actions=None, max_high_risk_per_session=None, max_mass_actions_per_session=DEFAULT_MAX_MASS_ACTIONS_PER_SESSION, max_recipients_per_mass=DEFAULT_MAX_RECIPIENTS, session_id=None)

In-process approval ledger for one agent run / publish session.

Enforces

• required token for high-risk actions
• single-use (or capped) consumption
• optional runaway budget (max high-risk successes per session)

This is the load-bearing reader for APPROVAL-GATE - not a flag file.

Source code in src/humanproof/closed_loop.py

def __init__(
    self,
    *,
    high_risk_actions: Iterable[str] | None = None,
    max_high_risk_per_session: int | None = None,
    max_mass_actions_per_session: int | None = DEFAULT_MAX_MASS_ACTIONS_PER_SESSION,
    max_recipients_per_mass: int = DEFAULT_MAX_RECIPIENTS,
    session_id: str | None = None,
) -> None:
    base = set(DEFAULT_HIGH_RISK_ACTIONS)
    if high_risk_actions is not None:
        base |= {_canonical_action(a) for a in high_risk_actions}
    self.high_risk_actions: frozenset[str] = frozenset(base)
    self.max_high_risk_per_session = max_high_risk_per_session
    self.max_mass_actions_per_session = max_mass_actions_per_session
    self.max_recipients_per_mass = max_recipients_per_mass
    self.session_id = session_id or secrets.token_hex(4)
    self._tokens: dict[str, ApprovalToken] = {}
    self._high_risk_passes: int = 0
    self._mass_action_passes: int = 0

classify(action)

Return high_risk or safe for action.

Source code in src/humanproof/closed_loop.py

def classify(self, action: str) -> str:
    """Return ``high_risk`` or ``safe`` for *action*."""
    canon = _canonical_action(action)
    if not canon:
        return "high_risk"  # empty action is never safe
    if canon in self.high_risk_actions:
        return "high_risk"
    # prefix match: "post:x_thread" → post
    head = canon.split(":", 1)[0]
    if head in self.high_risk_actions:
        return "high_risk"
    return "safe"

issue(action='*', *, ttl_seconds=3600.0, max_uses=1, issuer='owner', metadata=None)

Mint a human approval token. Call only from owner / HITL UI.

Source code in src/humanproof/closed_loop.py

def issue(
    self,
    action: str = "*",
    *,
    ttl_seconds: float | None = 3600.0,
    max_uses: int = 1,
    issuer: str = "owner",
    metadata: dict[str, Any] | None = None,
) -> ApprovalToken:
    """Mint a human approval token. Call only from owner / HITL UI."""
    if max_uses < 1:
        raise ApprovalError("max_uses must be >= 1")
    secret = secrets.token_urlsafe(24)
    token_id = hashlib.sha256(secret.encode()).hexdigest()[:16]
    now = time.time()
    expires = None if ttl_seconds is None else now + float(ttl_seconds)
    token = ApprovalToken(
        token_id=token_id,
        secret=secret,
        action=_canonical_action(action) if action != "*" else "*",
        issued_at=now,
        expires_at=expires,
        max_uses=max_uses,
        uses=0,
        issuer=issuer,
        metadata=dict(metadata or {}),
    )
    self._tokens[token_id] = token
    return token

record_mass_pass()

Increment mass-action counter after a successful mass gate.

Source code in src/humanproof/closed_loop.py

def record_mass_pass(self) -> None:
    """Increment mass-action counter after a successful mass gate."""
    self._mass_action_passes += 1

ApprovalToken(token_id, secret, action, issued_at, expires_at=None, max_uses=1, uses=0, issuer='owner', metadata=dict()) dataclass

Single-use (or multi-use) human-issued approval credential.

Agents must never create these for themselves in production; only a human (or an out-of-band owner control plane) calls :meth:ApprovalSession.issue.

fingerprint()

Public id for logs - not the secret.

Source code in src/humanproof/closed_loop.py

def fingerprint(self) -> str:
    """Public id for logs - not the secret."""
    return self.token_id

GateOutcome(ok, verdict, reason, exit_code, action=None, risk=None, human_required=False, token_id=None, approvals_remaining=None, recipient_count=0, mass_action_count=0) dataclass

Result of an approval or mass-action gate check.

Attributes:

Name	Type	Description
ok	bool	True only when the action may proceed.
verdict	str	PASS, FAIL, or FAIL_LOUD.
reason	str	Human-readable explanation (always non-empty).
exit_code	int	0 PASS, 1 FAIL (policy deny), 2 FAIL_LOUD (missing/empty).
action	str | None	Canonical action name that was gated.
risk	str | None	safe or high_risk.
human_required	bool	True when a human must issue a token.
token_id	str | None	Consumed or matched token id when present.
approvals_remaining	int | None	Budget remaining in the session after this check.
recipient_count	int	Recipients / targets in a mass-action gate.
mass_action_count	int	Mass actions already passed this session.

MotorFeatures(mean_speed, speed_std, noise_ratio, correction_rate, jerk_mean, jerk_std, max_speed, smoothness) dataclass

Statistical features extracted from an input trajectory.

Attributes:

Name	Type	Description
mean_speed	float	Mean speed in pixels/ms.
speed_std	float	Standard deviation of speed.
noise_ratio	float	std/mean speed. Humans ~0.4-0.8, AIs ~0.05-0.2.
correction_rate	float	Corrections per sample. Humans ~0.15-0.35.
jerk_mean	float	Mean absolute jerk.
jerk_std	float	Standard deviation of jerk.
max_speed	float	Maximum speed.
smoothness	float	Inverse of mean jerk (higher = smoother, more AI-like).

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/humanproof/scorer.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "mean_speed": self.mean_speed,
        "speed_std": self.speed_std,
        "noise_ratio": self.noise_ratio,
        "correction_rate": self.correction_rate,
        "jerk_mean": self.jerk_mean,
        "jerk_std": self.jerk_std,
        "max_speed": self.max_speed,
        "smoothness": self.smoothness,
    }

MotorScore(trajectory_id, features, human_score, ai_score, verdict, flags) dataclass

The scored result for a single trajectory.

Attributes:

Name	Type	Description
trajectory_id	str	ID of the scored trajectory.
features	MotorFeatures	Extracted motor features.
human_score	float	Probability of human input [0.0, 1.0].
ai_score	float	Probability of AI input (1.0 - human_score).
verdict	str	"human", "ai", or "uncertain".
flags	list[str]	List of flagged anomalies.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/humanproof/scorer.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "trajectory_id": self.trajectory_id,
        "features": self.features.to_dict(),
        "human_score": self.human_score,
        "ai_score": self.ai_score,
        "verdict": self.verdict,
        "flags": self.flags,
    }

MotorScorer

Score trajectories using threshold-based heuristics on motor features.

Human ranges: noise_ratio > 0.3, correction_rate > 0.1, smoothness < 5.0 AI ranges: noise_ratio < 0.15, correction_rate < 0.05, smoothness > 8.0

extract_features(traj)

Extract MotorFeatures from a trajectory.

Source code in src/humanproof/scorer.py

def extract_features(self, traj: InputTrajectory) -> MotorFeatures:
    """Extract MotorFeatures from a trajectory."""
    vp = traj.velocity_profile()
    jp = traj.jerk_profile()
    n = len(traj.samples)

    mean_speed = sum(vp) / len(vp) if vp else 0.0
    speed_variance = sum((v - mean_speed) ** 2 for v in vp) / len(vp) if vp else 0.0
    speed_std = math.sqrt(speed_variance)
    max_speed = max(vp) if vp else 0.0
    noise_ratio = traj.noise_ratio()
    correction_rate = traj.correction_count() / n if n > 0 else 0.0

    abs_jerks = [abs(j) for j in jp]
    jerk_mean = sum(abs_jerks) / len(abs_jerks) if abs_jerks else 0.0
    jerk_variance = (
        sum((j - jerk_mean) ** 2 for j in abs_jerks) / len(abs_jerks) if abs_jerks else 0.0
    )
    jerk_std = math.sqrt(jerk_variance)
    smoothness = 1.0 / (jerk_mean + 1e-9)

    return MotorFeatures(
        mean_speed=mean_speed,
        speed_std=speed_std,
        noise_ratio=noise_ratio,
        correction_rate=correction_rate,
        jerk_mean=jerk_mean,
        jerk_std=jerk_std,
        max_speed=max_speed,
        smoothness=smoothness,
    )

score(traj)

Score a trajectory and return a MotorScore.

Source code in src/humanproof/scorer.py

def score(self, traj: InputTrajectory) -> MotorScore:
    """Score a trajectory and return a MotorScore."""
    features = self.extract_features(traj)
    flags: list[str] = []
    human_score = 0.5

    # noise_ratio: < 0.15 → AI, > 0.3 → human
    if features.noise_ratio < 0.15:
        flags.append("low_noise_ratio")
        human_score -= 0.2
    elif features.noise_ratio > 0.3:
        human_score += 0.2

    # correction_rate: < 0.05 → AI, > 0.1 → human
    if features.correction_rate < 0.05:
        flags.append("low_correction_rate")
        human_score -= 0.15
    elif features.correction_rate > 0.1:
        human_score += 0.15

    # smoothness: > 8.0 → AI, < 5.0 → human
    if features.smoothness > 8.0:
        flags.append("high_smoothness")
        human_score -= 0.15
    elif features.smoothness < 5.0:
        human_score += 0.15

    # Clamp
    human_score = max(0.0, min(1.0, human_score))
    ai_score = round(1.0 - human_score, 2)

    if human_score > 0.65:
        verdict = "human"
    elif human_score < 0.35:
        verdict = "ai"
    else:
        verdict = "uncertain"

    return MotorScore(
        trajectory_id=traj.id,
        features=features,
        human_score=human_score,
        ai_score=ai_score,
        verdict=verdict,
        flags=flags,
    )

batch_score(trajs)

Score multiple trajectories and return one MotorScore per trajectory.

Parameters:

Name	Type	Description	Default
trajs	list[InputTrajectory]	List of InputTrajectory objects to score.	required

Returns:

Type	Description
list[MotorScore]	List of MotorScore objects in the same order as trajs.

Source code in src/humanproof/scorer.py

def batch_score(self, trajs: list[InputTrajectory]) -> list[MotorScore]:
    """Score multiple trajectories and return one MotorScore per trajectory.

    Args:
        trajs: List of InputTrajectory objects to score.

    Returns:
        List of MotorScore objects in the same order as *trajs*.
    """
    return [self.score(t) for t in trajs]

InputSample(dx, dy, dt, timestamp=0.0) dataclass

A single mouse/aim input sample.

Attributes:

Name	Type	Description
dx	float	X-axis delta (pixels).
dy	float	Y-axis delta (pixels).
dt	float	Time delta in milliseconds (must be > 0).
timestamp	float	Absolute timestamp in milliseconds.

InputTrajectory(samples, session_id='') dataclass

A sequence of input samples forming a trajectory.

Attributes:

Name	Type	Description
samples	list[InputSample]	Ordered list of input samples.
session_id	str	Optional session identifier string.
id	str	SHA-256[:16] fingerprint computed from session_id, length, first/last sample.

velocity_profile()

Compute speed at each sample step (pixels/ms).

Source code in src/humanproof/trajectory.py

def velocity_profile(self) -> list[float]:
    """Compute speed at each sample step (pixels/ms)."""
    return [math.sqrt(s.dx**2 + s.dy**2) / s.dt for s in self.samples]

acceleration_profile()

Compute acceleration at each step (pixels/ms^2). Length = n-1.

Source code in src/humanproof/trajectory.py

def acceleration_profile(self) -> list[float]:
    """Compute acceleration at each step (pixels/ms^2). Length = n-1."""
    vp = self.velocity_profile()
    return [vp[i + 1] - vp[i] for i in range(len(vp) - 1)]

jerk_profile()

Compute jerk at each step (pixels/ms^3). Length = n-2.

Source code in src/humanproof/trajectory.py

def jerk_profile(self) -> list[float]:
    """Compute jerk at each step (pixels/ms^3). Length = n-2."""
    ap = self.acceleration_profile()
    return [ap[i + 1] - ap[i] for i in range(len(ap) - 1)]

correction_count()

Count direction reversals (velocity sign flips in x or y).

Source code in src/humanproof/trajectory.py

def correction_count(self) -> int:
    """Count direction reversals (velocity sign flips in x or y)."""
    count = 0
    for i in range(1, len(self.samples)):
        prev = self.samples[i - 1]
        curr = self.samples[i]
        x_flip = prev.dx != 0 and curr.dx != 0 and (prev.dx > 0) != (curr.dx > 0)
        y_flip = prev.dy != 0 and curr.dy != 0 and (prev.dy > 0) != (curr.dy > 0)
        if x_flip or y_flip:
            count += 1
    return count

noise_ratio()

Compute std(velocity) / mean(abs(velocity)). Returns 0.0 if mean is 0.

Source code in src/humanproof/trajectory.py

def noise_ratio(self) -> float:
    """Compute std(velocity) / mean(abs(velocity)). Returns 0.0 if mean is 0."""
    vp = self.velocity_profile()
    if not vp:
        return 0.0
    mean_v = sum(vp) / len(vp)
    if mean_v == 0:
        return 0.0
    variance = sum((v - mean_v) ** 2 for v in vp) / len(vp)
    std_v = math.sqrt(variance)
    return std_v / mean_v

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/humanproof/trajectory.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "session_id": self.session_id,
        "samples": [
            {"dx": s.dx, "dy": s.dy, "dt": s.dt, "timestamp": s.timestamp} for s in self.samples
        ],
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/humanproof/trajectory.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> InputTrajectory:
    """Deserialize from a dict produced by to_dict()."""
    samples = [
        InputSample(
            dx=float(s["dx"]),
            dy=float(s["dy"]),
            dt=float(s["dt"]),
            timestamp=float(s.get("timestamp", 0.0)),
        )
        for s in d["samples"]
    ]
    return cls(samples=samples, session_id=d.get("session_id", ""))

batch_score(trajectories, scorer=None)

Score a list of trajectories and return aggregated BatchScoreResult.

Source code in src/humanproof/batch.py

def batch_score(
    trajectories: list[InputTrajectory], scorer: MotorScorer | None = None
) -> BatchScoreResult:
    """Score a list of trajectories and return aggregated BatchScoreResult."""
    if scorer is None:
        scorer = MotorScorer()
    scores = scorer.batch_score(trajectories)
    human_count = sum(1 for s in scores if s.verdict == "human")
    ai_count = sum(1 for s in scores if s.verdict == "ai")
    uncertain_count = sum(1 for s in scores if s.verdict == "uncertain")
    mean_human_score = sum(s.human_score for s in scores) / len(scores) if scores else 0.0
    flagged = [s.trajectory_id for s in scores if s.verdict == "ai"]
    total = len(scores)
    summary = (
        f"Scored {total} trajectories: {human_count} human, {ai_count} AI, "
        f"{uncertain_count} uncertain. Mean human score: {mean_human_score:.2f}."
    )
    return BatchScoreResult(
        scores=scores,
        human_count=human_count,
        ai_count=ai_count,
        uncertain_count=uncertain_count,
        mean_human_score=mean_human_score,
        flagged_trajectories=flagged,
        summary=summary,
    )

score_from_csv(csv_path)

Load trajectories from CSV (columns: trajectory_id,t,x,y,button) and score them.

Each unique trajectory_id forms one InputTrajectory. Rows are sorted by t. x,y are treated as absolute positions; dx/dy are computed from consecutive rows. button column is ignored (reserved for future use).

Source code in src/humanproof/batch.py

def score_from_csv(csv_path: Path) -> BatchScoreResult:
    """Load trajectories from CSV (columns: trajectory_id,t,x,y,button) and score them.

    Each unique trajectory_id forms one InputTrajectory. Rows are sorted by t.
    x,y are treated as absolute positions; dx/dy are computed from consecutive rows.
    button column is ignored (reserved for future use).
    """
    csv_path = Path(csv_path)
    rows_by_id: dict[str, list[dict[str, Any]]] = {}
    with csv_path.open(newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            tid = row["trajectory_id"]
            rows_by_id.setdefault(tid, []).append(row)

    trajectories: list[InputTrajectory] = []
    for tid, rows in rows_by_id.items():
        rows.sort(key=lambda r: float(r["t"]))
        samples: list[InputSample] = []
        prev_x: float | None = None
        prev_y: float | None = None
        prev_t: float | None = None
        for r in rows:
            x = float(r["x"])
            y = float(r["y"])
            t = float(r["t"])
            if prev_x is None:
                prev_x, prev_y, prev_t = x, y, t
                continue
            dx = x - prev_x
            dy = y - (prev_y or 0.0)
            dt = t - (prev_t or 0.0)
            if dt <= 0:
                dt = 1.0
            samples.append(InputSample(dx=dx, dy=dy, dt=dt, timestamp=t))
            prev_x, prev_y, prev_t = x, y, t
        if len(samples) >= 1:
            traj = InputTrajectory(samples=samples, session_id=tid)
            trajectories.append(traj)

    return batch_score(trajectories)

apply_calibration(scorer, calibration)

Return a CalibratedMotorScorer with calibrated thresholds.

Source code in src/humanproof/calibration.py

def apply_calibration(scorer: MotorScorer, calibration: CalibrationResult) -> CalibratedMotorScorer:
    """Return a CalibratedMotorScorer with calibrated thresholds."""
    return CalibratedMotorScorer(
        calibration.optimal_noise_threshold,
        calibration.optimal_correction_threshold,
    )

calibrate(human_trajectories, ai_trajectories)

Find optimal decision thresholds via grid search over noise and correction thresholds.

Source code in src/humanproof/calibration.py

def calibrate(
    human_trajectories: list[InputTrajectory],
    ai_trajectories: list[InputTrajectory],
) -> CalibrationResult:
    """Find optimal decision thresholds via grid search over noise and correction thresholds."""
    noise_candidates = [0.05, 0.10, 0.15, 0.20, 0.25, 0.30]
    correction_candidates = [0.03, 0.05, 0.08, 0.10, 0.12, 0.15]

    best_accuracy = -1.0
    best_noise = 0.15
    best_correction = 0.05
    best_human_precision = 0.0
    best_ai_precision = 0.0

    for noise_t in noise_candidates:
        for corr_t in correction_candidates:
            acc, hp, ap = _evaluate(human_trajectories, ai_trajectories, noise_t, corr_t)
            if acc > best_accuracy:
                best_accuracy = acc
                best_noise = noise_t
                best_correction = corr_t
                best_human_precision = hp
                best_ai_precision = ap

    return CalibrationResult(
        optimal_noise_threshold=best_noise,
        optimal_correction_threshold=best_correction,
        accuracy=best_accuracy,
        human_precision=best_human_precision,
        ai_precision=best_ai_precision,
    )

assert_approved(action, token=None, **kwargs)

Gate and raise :class:ApprovalError unless outcome is ok.

Source code in src/humanproof/closed_loop.py

def assert_approved(
    action: str,
    token: ApprovalToken | str | None = None,
    **kwargs: Any,
) -> GateOutcome:
    """Gate and raise :class:`ApprovalError` unless outcome is ok."""
    outcome = gate_approval(action, token, **kwargs)
    if not outcome.ok:
        raise ApprovalError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_mass_action_ok(action, recipients=None, **kwargs)

Raise :class:ApprovalError unless :func:gate_mass_action is ok.

Source code in src/humanproof/closed_loop.py

def assert_mass_action_ok(
    action: str,
    recipients: Sequence[str] | None = None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ApprovalError` unless :func:`gate_mass_action` is ok."""
    outcome = gate_mass_action(action, recipients, **kwargs)
    if not outcome.ok:
        raise ApprovalError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_approval(action, token=None, *, session=None, secret=None, consume=True)

Gate a proposed action: high-risk requires a valid human token.

Parameters:

Name	Type	Description	Default
action	str	Proposed action name (e.g. post, db_wipe, score).	required
token	ApprovalToken | str | None	:class:ApprovalToken or token_id string from :meth:ApprovalSession.issue.	None
session	ApprovalSession | None	Session ledger; created empty if omitted (then only safe actions pass).	None
secret	str | None	Optional secret if token is a token_id string (constant-time check).	None
consume	bool	If True (default), successful high-risk checks increment token uses.	True

Returns:

Type	Description
GateOutcome	class:GateOutcome. Callers should refuse the action unless ok.

Source code in src/humanproof/closed_loop.py

def gate_approval(
    action: str,
    token: ApprovalToken | str | None = None,
    *,
    session: ApprovalSession | None = None,
    secret: str | None = None,
    consume: bool = True,
) -> GateOutcome:
    """Gate a proposed action: high-risk requires a valid human token.

    Args:
        action: Proposed action name (e.g. ``post``, ``db_wipe``, ``score``).
        token: :class:`ApprovalToken` or token_id string from :meth:`ApprovalSession.issue`.
        session: Session ledger; created empty if omitted (then only safe actions pass).
        secret: Optional secret if *token* is a token_id string (constant-time check).
        consume: If True (default), successful high-risk checks increment token uses.

    Returns:
        :class:`GateOutcome`. Callers should refuse the action unless ``ok``.
    """
    sess = session or ApprovalSession()
    remaining = sess.remaining_budget()
    canon = _canonical_action(action)

    if not canon:
        return _fail_loud(
            "empty action - refuse (APPROVAL-GATE: nothing to approve)",
            action=action or "",
            risk="high_risk",
            remaining=remaining,
        )

    risk = sess.classify(canon)

    if risk == "safe":
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=f"safe action {canon!r} - no approval required",
            exit_code=0,
            action=canon,
            risk="safe",
            human_required=False,
            token_id=None,
            approvals_remaining=remaining,
        )

    # High-risk: token required
    if token is None:
        return _fail_loud(
            f"high-risk action {canon!r} requires owner approval token "
            f"(APPROVAL-GATE / auto-post without owner)",
            action=canon,
            risk="high_risk",
            remaining=remaining,
        )

    # Resolve token object
    tok: ApprovalToken | None
    if isinstance(token, ApprovalToken):
        tok = token
        # Prefer session registry if present (authoritative uses counter)
        registered = sess.peek(token.token_id)
        if registered is not None:
            tok = registered
    else:
        tok = sess.peek(str(token))
        if tok is None:
            return _fail(
                f"unknown approval token_id {token!r} for action {canon!r}",
                action=canon,
                risk="high_risk",
                remaining=remaining,
            )

    if secret is not None and not secrets.compare_digest(tok.secret, secret):
        return _fail(
            "approval secret mismatch",
            action=canon,
            risk="high_risk",
            token_id=tok.token_id,
            remaining=remaining,
        )

    if tok.expired:
        return _fail(
            f"approval token {tok.token_id} expired",
            action=canon,
            risk="high_risk",
            token_id=tok.token_id,
            remaining=remaining,
        )

    if tok.exhausted:
        return _fail(
            f"approval token {tok.token_id} exhausted ({tok.uses}/{tok.max_uses} uses)",
            action=canon,
            risk="high_risk",
            token_id=tok.token_id,
            remaining=remaining,
        )

    if not tok.matches_action(canon):
        return _fail(
            f"token {tok.token_id} issued for action {tok.action!r}, not {canon!r}",
            action=canon,
            risk="high_risk",
            token_id=tok.token_id,
            remaining=remaining,
        )

    # Runaway budget (Guardian / AgentWatch class)
    if (
        sess.max_high_risk_per_session is not None
        and sess.high_risk_pass_count >= sess.max_high_risk_per_session
    ):
        return _fail(
            f"session high-risk budget exhausted "
            f"({sess.high_risk_pass_count}/{sess.max_high_risk_per_session}) "
            f"- re-approval required (runaway guard)",
            action=canon,
            risk="high_risk",
            token_id=tok.token_id,
            remaining=0,
        )

    if consume:
        tok.uses += 1
        sess._tokens[tok.token_id] = tok
        sess._high_risk_passes += 1

    remaining_after = sess.remaining_budget()
    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"approved {canon!r} via token {tok.token_id} (issuer={tok.issuer})",
        exit_code=0,
        action=canon,
        risk="high_risk",
        human_required=False,
        token_id=tok.token_id,
        approvals_remaining=remaining_after,
    )

gate_mass_action(action, recipients=None, *, token=None, session=None, secret=None, max_recipients=None, require_inventory=True, consume=True)

Block unattended bulk email/delete (OpenClaw mass-email class).

Load-bearing controls:

1. Classify - action must be mass/bulk class (or still go through :func:gate_approval if high-risk single).
2. Inventory - named recipients/targets required when require_inventory (empty list → FAIL_LOUD).
3. Bulk limit - recipient count over session/default max without a valid approval token → FAIL_LOUD.
4. Approval - always requires human token via :func:gate_approval for mass actions (never unattended).
5. Session mass budget - max mass actions per session (AgentWatch class).

Parameters:

Name	Type	Description	Default
action	str	e.g. mass_email, send_email, mass_delete.	required
recipients	Sequence[str] | None	Explicit list of email addresses / targets / ids.	None
token	ApprovalToken | str | None	Human approval token (required for mass actions).	None
session	ApprovalSession | None	Approval session (mass + high-risk budgets).	None
secret	str | None	Optional secret when token is a token_id string.	None
max_recipients	int | None	Override max recipients (default session or 50).	None
require_inventory	bool	If True, empty recipients FAIL_LOUD.	True
consume	bool	Pass-through to :func:gate_approval on success path.	True

Source code in src/humanproof/closed_loop.py

def gate_mass_action(
    action: str,
    recipients: Sequence[str] | None = None,
    *,
    token: ApprovalToken | str | None = None,
    session: ApprovalSession | None = None,
    secret: str | None = None,
    max_recipients: int | None = None,
    require_inventory: bool = True,
    consume: bool = True,
) -> GateOutcome:
    """Block unattended bulk email/delete (OpenClaw mass-email class).

    Load-bearing controls:

    1. **Classify** - action must be mass/bulk class (or still go through
       :func:`gate_approval` if high-risk single).
    2. **Inventory** - named recipients/targets required when
       ``require_inventory`` (empty list → FAIL_LOUD).
    3. **Bulk limit** - recipient count over session/default max without a
       valid approval token → FAIL_LOUD.
    4. **Approval** - always requires human token via :func:`gate_approval`
       for mass actions (never unattended).
    5. **Session mass budget** - max mass actions per session (AgentWatch class).

    Args:
        action: e.g. ``mass_email``, ``send_email``, ``mass_delete``.
        recipients: Explicit list of email addresses / targets / ids.
        token: Human approval token (required for mass actions).
        session: Approval session (mass + high-risk budgets).
        secret: Optional secret when *token* is a token_id string.
        max_recipients: Override max recipients (default session or 50).
        require_inventory: If True, empty recipients FAIL_LOUD.
        consume: Pass-through to :func:`gate_approval` on success path.
    """
    canon = _canonical_action(action)
    sess = session or ApprovalSession()
    recips = _recipient_list(recipients)
    n = len(recips)
    mass_count = sess.mass_action_pass_count
    limit = (
        max_recipients
        if max_recipients is not None
        else getattr(sess, "max_recipients_per_mass", DEFAULT_MAX_RECIPIENTS)
    )

    if not canon:
        return _fail_loud(
            "MASS-EMAIL: empty action - cannot gate phantom bulk side effect",
            action=None,
            risk="high_risk",
            recipient_count=n,
            mass_action_count=mass_count,
        )

    if not is_mass_action(canon):
        # Non-mass: fall through to standard approval gate.
        return gate_approval(canon, token, session=sess, secret=secret, consume=consume)

    if require_inventory and n == 0:
        return _fail_loud(
            f"MASS-EMAIL/OpenClaw: mass action {canon!r} without recipient "
            f"inventory - agent must name targets before bulk send/delete",
            action=canon,
            risk="high_risk",
            recipient_count=0,
            mass_action_count=mass_count,
        )

    if n > limit:
        return _fail_loud(
            f"MASS-EMAIL/OpenClaw: recipient_count={n} exceeds max={limit} "
            f"for {canon!r} - refuse bulk external side effect "
            f"(public: OpenClaw mass email delete class)",
            action=canon,
            risk="high_risk",
            recipient_count=n,
            mass_action_count=mass_count,
        )

    # Session mass budget (runaway bulk loops).
    if (
        sess.max_mass_actions_per_session is not None
        and sess.mass_action_pass_count >= sess.max_mass_actions_per_session
    ):
        return _fail(
            f"MASS-EMAIL: session mass-action budget exhausted "
            f"({sess.mass_action_pass_count}/{sess.max_mass_actions_per_session}) "
            f"- re-approval required (AgentWatch/Guardian class)",
            action=canon,
            risk="high_risk",
            recipient_count=n,
            mass_action_count=mass_count,
        )

    # Always require human approval for mass actions.
    auth = gate_approval(canon, token, session=sess, secret=secret, consume=consume)
    if not auth.ok:
        return GateOutcome(
            ok=False,
            verdict=auth.verdict,
            reason=(f"MASS-EMAIL/OpenClaw: {auth.reason} (recipients={n} action={canon!r})"),
            exit_code=auth.exit_code,
            action=canon,
            risk="high_risk",
            human_required=True,
            token_id=auth.token_id,
            approvals_remaining=auth.approvals_remaining,
            recipient_count=n,
            mass_action_count=mass_count,
        )

    if consume:
        sess.record_mass_pass()

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"mass action authorised: action={canon!r} recipients={n} "
            f"token={auth.token_id} mass_count={sess.mass_action_pass_count}"
        ),
        exit_code=0,
        action=canon,
        risk="high_risk",
        human_required=False,
        token_id=auth.token_id,
        approvals_remaining=auth.approvals_remaining,
        recipient_count=n,
        mass_action_count=sess.mass_action_pass_count,
    )

is_mass_action(action)

True if action is a bulk external side effect (email/delete class).

Source code in src/humanproof/closed_loop.py

def is_mass_action(action: str) -> bool:
    """True if *action* is a bulk external side effect (email/delete class)."""
    canon = _canonical_action(action)
    if not canon:
        return False
    if canon in DEFAULT_MASS_ACTIONS:
        return True
    head = canon.split(":", 1)[0]
    return head in DEFAULT_MASS_ACTIONS

require_token_for(action, session)

Convenience: gate without a token - always FAIL_LOUD for high-risk.

Useful in tests and CI to prove the unattended path is blocked.

Source code in src/humanproof/closed_loop.py

def require_token_for(
    action: str,
    session: ApprovalSession,
) -> GateOutcome:
    """Convenience: gate without a token - always FAIL_LOUD for high-risk.

    Useful in tests and CI to prove the unattended path is blocked.
    """
    return gate_approval(action, token=None, session=session)

analyze_session(session_id, trajectories)

Analyze a gaming session consisting of multiple trajectories.

Source code in src/humanproof/session.py

def analyze_session(session_id: str, trajectories: list[InputTrajectory]) -> SessionAnalysis:
    """Analyze a gaming session consisting of multiple trajectories."""
    if not trajectories:
        raise ValueError("trajectories must not be empty")
    scorer = MotorScorer()
    result = batch_score(trajectories, scorer=scorer)

    human_scores = [s.human_score for s in result.scores]
    score_over_time = [(i, human_scores[i]) for i in range(len(human_scores))]

    shift_idx = detect_shift(human_scores)
    behavioral_shift_detected = shift_idx is not None

    mean_score = result.mean_human_score

    if behavioral_shift_detected:
        verdict = "behavioral_shift"
        risk_level = "high"
    elif mean_score >= 0.65:
        verdict = "consistent_human"
        risk_level = "low"
    elif mean_score <= 0.35:
        verdict = "consistent_ai"
        risk_level = "high"
    else:
        verdict = "uncertain"
        risk_level = "medium"

    return SessionAnalysis(
        session_id=session_id,
        trajectory_count=len(trajectories),
        mean_human_score=mean_score,
        score_over_time=score_over_time,
        behavioral_shift_detected=behavioral_shift_detected,
        shift_at_index=shift_idx,
        verdict=verdict,
        risk_level=risk_level,
    )

detect_shift(scores, window=3, threshold=0.3)

Detect the index where mean score changed by more than threshold.

Compares the mean of the window before vs after each index. Returns the first index where the shift exceeds the threshold.

Source code in src/humanproof/session.py

def detect_shift(scores: list[float], window: int = 3, threshold: float = 0.3) -> int | None:
    """Detect the index where mean score changed by more than threshold.

    Compares the mean of the window before vs after each index.
    Returns the first index where the shift exceeds the threshold.
    """
    n = len(scores)
    if n < window * 2:
        return None
    for i in range(window, n - window + 1):
        before = scores[max(0, i - window) : i]
        after = scores[i : i + window]
        mean_before = sum(before) / len(before)
        mean_after = sum(after) / len(after)
        if abs(mean_after - mean_before) > threshold:
            return i
    return None