Security, usage and audit

Three optional components guard the pipeline. Each is off until you attach it.

Component	Crate	Attach with	Blocks with
Security rules	beam_security	with_security / set_security	BEM-061 (403)
Usage quotas	beam_usage	set_usage(guard, tracker, plan)	BEM-060 (403)
Audit events	beam_audit	set_audit	never blocks

worker/src/main.rs

use std::sync::Arc;

use beam_audit::{AuditLogger, StdoutSink};
use beam_security::{SecurityManager, SecurityPolicy};
use beam_usage::{InMemoryUsageStore, UsageGuard, UsagePlan, UsageTracker};

beam.set_security(Arc::new(SecurityManager::new(SecurityPolicy::default())));

let tracker = Arc::new(UsageTracker::new(Arc::new(InMemoryUsageStore::new())));
beam.set_usage(Arc::new(UsageGuard::new(tracker.clone())), tracker, UsagePlan::unlimited("default"));

beam.set_audit(Arc::new(AuditLogger::new(vec![Arc::new(StdoutSink)])));

Security

SecurityManager evaluates a rule base with the river_reed GRL engine. Regular-expression scorers in Rust turn the content into facts, such as an injection score or a secrets score, and the rules decide on those facts.

The pipeline checks two points:

Stage	Content checked
BeforeInput	The text parts of the request, before anything is saved
BeforeOutput	The text of the agent's reply, before it is saved or returned

A verdict that denies (Deny, Terminate, Quarantine) or asks for approval (RequireApproval) blocks the request with BEM-061, and the decision's user_message goes into the error data. Every check is audited as a security_event with the stage, action, score, severity and matched rules.

SecurityManager::new(policy).with_enforce(false) evaluates and audits without blocking, which is useful while you tune rules. The default, DisabledSecurityManager, allows everything.

Shipped rules

Rule	Stage	Fires when	Verdict
PI-001	input	Injection score ≥ 0.7 / between 0.4 and 0.7	Deny / Warn
PI-008	input	Role-override score ≥ 0.4	Deny
OUT-001	output	Secrets score ≥ 0.4	Deny
OUT-002	output	PII score ≥ 0.4	Warn
TOOL-001	tool	Tool is globally denied, or not in a non-empty allowlist	Deny
HITL-001	tool	Tool needs approval under the policy	RequireApproval
MEM-001	memory	Sensitive content in a memory save	Deny
MEM-005	memory	Payload confidence ≥ 0.7	Warn
RUN-001 to RUN-005	runtime	Steps, time, tokens or recursion over the policy limits, or a loop	Deny (Warn near the step limit)

SecurityPolicy holds the tool lists (global_denied_tools, global_allowed_tools, tool_policies), the approval setting (hitl_config), and the runtime limits: max_steps 50, max_tokens 128 000, max_recursion 10, max_timeout 30 seconds.

You can call the manager yourself for the other stages:

use beam_security::{SecurityManagement, SecurityRequest, SecurityStage};
use serde_json::json;

let payload = json!({"content": "hello there"}).as_object().cloned().unwrap_or_default();
let decision = manager.check(&SecurityRequest::new(SecurityStage::BeforeInput).with_payload(payload)).await;
if decision.denies() || decision.requires_hitl() {
    // block
}

Current limitations

Do not rely on the shipped rules for prompt injection

Each scorer caps at 0.5, so PI-001 never blocks, and common phrasings such as "ignore all previous instructions" score below the 0.4 warning threshold. Secrets and PII patterns also need a higher-weighted match to reach 0.4: an AWS-style key is blocked, a plain password= is not.

• On a streaming request, the agent's frames, including the done frame with the full reply, are published before the BeforeOutput check runs. A blocked reply has already reached subscribers; only the returned result and the stored message are stopped.
• The pipeline runs only the BeforeInput and BeforeOutput stages. Tool, memory and runtime rules run only when you call the manager yourself.
• If the rule base fails to load or evaluate, the decision is Allow. Check is_ready() at startup.
• The shipped rules produce Allow, Warn, RequireApproval and Deny only. Content is never sanitized or masked.

Usage quotas

A UsagePlan sets limits per user. The guard checks them before the agent runs, and the tracker records usage after a successful run.

use beam_usage::{OveragePolicy, UsageLimits, UsagePlan};

let plan = UsagePlan {
    name: "basic".into(),
    limits: UsageLimits {
        max_requests_per_day: Some(200),
        max_tokens_per_day: Some(100_000),
        allowed_agents: Some(vec!["assistant".into()]),
        ..Default::default()
    },
    overage_policy: OveragePolicy::HardBlock,
};

Limit	Denies when
max_requests_per_day	Today's request count has reached the limit
max_tokens_per_day	Today's token count has reached the limit
allowed_agents	The agent is not listed
allowed_modalities	The request uses a modality that is not listed
allowed_models	model_override names a model that is not listed

Days are UTC. Requests without auth_context.user_id are not checked or counted. One plan applies to every user; UsagePlan::unlimited(name) sets no limits.

Current limitations

• Only OveragePolicy::HardBlock blocks. SoftWarn and Throttle let the request through.
• max_tokens_per_request is not enforced.
• RigAgentRunner reports zero tokens and no model_used, so token counts stay at 0 for its runs and max_tokens_per_day never blocks them. Request limits still work.
• InMemoryUsageStore is the only store, so counters reset on restart and are not shared between workers. Implement UsageStore for shared counters.

Audit

AuditLogger sends each AuditEvent to its sinks in order. The pipeline logs these actions:

Action	When
request_received	Every request, first
quota_exceeded	The usage guard blocked the request
security_event	Every security check
agent_resumed	A resume turn resolved its actions
agent_completed / agent_interrupted	The run finished
agent_failed	The run failed

Events carry timestamp, action, agent_name, user_id, session_id, run_id, status (ok, error, blocked or flagged), error, and metadata.

Sink	Behaviour
StdoutSink	Prints one JSON line per event
InMemorySink	Keeps events in memory; read them with events()

Write your own sink to store events elsewhere:

use async_trait::async_trait;
use beam_audit::{AuditEvent, AuditSink};

struct TracingSink;

#[async_trait]
impl AuditSink for TracingSink {
    async fn write(&self, event: &AuditEvent) {
        tracing::info!(action = ?event.action, status = %event.status, "beam audit");
    }
}

The schema has an agent_event table for audit events, but no shipped sink writes to it yet.