Session-Aware Guards

The session journal

Source: crates/platform/chio-http-session/src/lib.rs. The journal is a thread-safe, hash-chained record of tool-call entries stored in capacity-bounded rings:

pub struct SessionJournal {
    inner: Mutex<JournalInner>,
    session_id: String,
}

struct JournalInner {
    // Capacity-bounded rings, not unbounded Vecs.
    entries: chio_bounded::Ring<JournalEntry>,
    tool_sequence: chio_bounded::Ring<String>,
    // Running fold over evicted entry hashes, so the chain stays
    // committed after the ring drops its prefix.
    evicted_head_hash: Option<String>,
    // Monotonic, independent of the ring length so sequence numbers
    // never repeat after eviction.
    next_sequence: u64,
    data_flow: CumulativeDataFlow,
    // Cumulative counts; the DISTINCT-key set is bounded fail-closed.
    tool_counts: HashMap<String, u64>,
    tool_counts_cap: usize,
    // O(1) cumulative streak (last tool + consecutive count) that
    // survives ring eviction.
    current_streak_tool: Option<String>,
    current_streak_len: u64,
}

pub struct CumulativeDataFlow {
    pub total_bytes_read: u64,
    pub total_bytes_written: u64,
    pub total_invocations: u64,
    pub max_delegation_depth: u32,
}

entries and tool_sequence are chio_bounded::Rings, defaulting to caps sourced from chio_kernel::MemoryBudgetConfig. When the ring evicts an entry its hash folds into evicted_head_hash, so integrity verification still commits to the dropped prefix. The two pieces of state the session-aware guards read cumulatively — tool_counts and the current_streak_tool / current_streak_len pair — survive that eviction, and tool_counts is itself distinct-key bounded (tool_counts_cap), dropping a newly-seen name fail-closed once full.

The guards read the journal through one snapshot instead of separate per-field accessors:

• snapshot() -> Result<SessionJournalSnapshot, SessionJournalError> takes the lock once and returns data flow, tool sequence, tool counts, the streak pair, and the head hash together, so no guard samples a torn read across fields. SessionJournalSnapshot carries session_id, entry_count, head_hash, data_flow, tool_sequence, tool_counts, current_streak_tool, and current_streak_len.
• record(RecordParams) -> Result<u64, SessionJournalError> appends a hash-chained entry and returns its sequence number. RecordParams carries tool_name, server_id, agent_id, bytes_read, bytes_written, delegation_depth, and allowed.
• data_flow() and tool_sequence() still exist as public accessors, but neither DataFlowGuard nor BehavioralSequenceGuard calls them anymore — both now go through snapshot().

The cumulative counters use saturating_add on every record, so each running total clamps at u64::MAX rather than wrapping. The guards take Arc<SessionJournal> at construction so they share the underlying state without owning it.

DataFlowGuard

Source: crates/guards/chio-guards/src/data_flow.rs. Guard name: data-flow (data_flow.rs:52). Reads cumulative bytes from the journal and denies once any configured ceiling is reached.

Struct

#[derive(Clone, Debug, Default)]
pub struct DataFlowConfig {
    pub max_bytes_read: Option<u64>,
    pub max_bytes_written: Option<u64>,
    pub max_bytes_total: Option<u64>,
}

pub struct DataFlowGuard {
    journal: Arc<SessionJournal>,
    config: DataFlowConfig,
}

Default on DataFlowConfig sets every ceiling to None: a default guard never denies. Per-knob defaults:

Algorithm

The evaluate body (data_flow.rs:57-88) reads cumulative data flow from a snapshot(), not the older data_flow() accessor:

fn evaluate(&self, _ctx: &GuardContext) -> Result<GuardDecision, KernelError> {
    let snapshot = self.journal.snapshot().map_err(|e| {
        KernelError::Internal(format!("data-flow guard journal error (fail-closed): {e}"))
    })?;
    let flow = snapshot.data_flow;

    if let Some(max_read) = self.config.max_bytes_read {
        if flow.total_bytes_read >= max_read {
            return Ok(GuardDecision::deny(Vec::new()));
        }
    }

    if let Some(max_written) = self.config.max_bytes_written {
        if flow.total_bytes_written >= max_written {
            return Ok(GuardDecision::deny(Vec::new()));
        }
    }

    if let Some(max_total) = self.config.max_bytes_total {
        let total = flow
            .total_bytes_read
            .saturating_add(flow.total_bytes_written);
        if total >= max_total {
            return Ok(GuardDecision::deny(Vec::new()));
        }
    }

    Ok(GuardDecision::allow())
}

The comparison is inclusive: a session that has already read exactly max_bytes_read denies the next call. The guard does not pre-charge the in-flight request, so the arithmetic only sees what prior callers wrote into the journal. The action enum is ignored: even invocations with zero reported bytes execute the three checks before returning Allow.

u64 ceiling

Both CumulativeDataFlow counters and max_bytes_total are u64. The journal's saturating_add updates clamp at u64::MAX = 18_446_744_073_709_551_615 bytes (about 16 EB, 18.4 quintillion bytes). At any realistic web scale this ceiling is unreachable. Saturation prevents a misconfigured journal from wrapping a counter to zero and re-allowing a terminated session.

Failure modes

• Journal lock poisoned :: SessionJournalError::LockPoisoned is mapped by map_err as KernelError::Internal("data-flow guard journal error (fail-closed): {e}"). The kernel reads Err(_) from a guard as a denial.
• Saturated counter :: deny stays deny. Once any total reaches its ceiling, every subsequent call denies until the session is replaced.

BehavioralSequenceGuard

Source: crates/guards/chio-guards/src/behavioral_sequence.rs. Guard name: behavioral-sequence (behavioral_sequence.rs:58). Enforces tool-ordering rules over the journal's tool sequence.

Struct

#[derive(Clone, Debug, Default)]
pub struct SequencePolicy {
    pub required_predecessors: HashMap<String, HashSet<String>>,
    pub forbidden_transitions: Vec<(String, String)>,
    pub max_consecutive: Option<u32>,
    pub required_first_tool: Option<String>,
}

pub struct BehavioralSequenceGuard {
    journal: Arc<SessionJournal>,
    policy: SequencePolicy,
}

Configuration

Algorithm

evaluate takes one snapshot() and checks four rules against cumulative journal fields — not against a streak re-derived from the bounded tool_sequence ring tail. Body, verbatim (behavioral_sequence.rs:63-147):

fn evaluate(&self, ctx: &GuardContext) -> Result<GuardDecision, KernelError> {
    let tool_name = &ctx.request.tool_name;

    let snapshot = self.journal.snapshot().map_err(|e| {
        KernelError::Internal(format!(
            "behavioral-sequence guard journal error (fail-closed): {e}"
        ))
    })?;

    // Required first tool: "nothing recorded yet" is cumulative, so read the
    // O(1) last-tool field, not the bounded tool_sequence ring.
    if snapshot.current_streak_tool.is_none() {
        if let Some(ref required_first) = self.policy.required_first_tool {
            if tool_name != required_first {
                return Ok(GuardDecision::deny(Vec::new()));
            }
        }
    }

    // Required predecessors: "ever invoked" is cumulative, so read tool_counts
    // (survives ring eviction; distinct-key bounded fail-closed).
    if let Some(required) = self.policy.required_predecessors.get(tool_name) {
        for req in required {
            if !snapshot.tool_counts.contains_key(req) {
                return Ok(GuardDecision::deny(Vec::new()));
            }
        }
    }

    // Forbidden transitions: the "last tool" is the cumulative streak tool.
    if let Some(last_tool) = snapshot.current_streak_tool.as_deref() {
        for (from, to) in &self.policy.forbidden_transitions {
            if last_tool == from && tool_name == to {
                return Ok(GuardDecision::deny(Vec::new()));
            }
        }
    }

    // Max consecutive: read the O(1) cumulative streak counter, which survives
    // ring eviction. A different requested tool starts a fresh streak.
    if let Some(max_consec) = self.policy.max_consecutive {
        let prior_streak =
            if snapshot.current_streak_tool.as_deref() == Some(tool_name.as_str()) {
                snapshot.current_streak_len
            } else {
                0
            };
        if prior_streak >= u64::from(max_consec) {
            return Ok(GuardDecision::deny(Vec::new()));
        }
    }

    Ok(GuardDecision::allow())
}

BehavioralProfileGuard

Source: crates/guards/chio-guards/src/behavioral_profile.rs. Guard name: behavioral-profile (behavioral_profile.rs:213). Computes anomaly signals against a per-agent EMA baseline. The verdict path is advisory: even when the sample is anomalous, evaluate returns GuardDecision::allow() (behavioral_profile.rs:355-365).

Defaults

Every default constant is declared at the top of the module (behavioral_profile.rs:46-54):

pub const DEFAULT_EMA_ALPHA: f64 = 0.2;
pub const DEFAULT_SIGMA_THRESHOLD: f64 = 2.0;
pub const DEFAULT_WINDOW_SECS: u64 = 60;
pub const DEFAULT_BASELINE_MIN_WINDOWS: u64 = 3;

EmaBaselineState

The baseline state is shared with the operator-report module, now a submodule directory (operator_report/behavioral_analysis.rs:158-198):

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct EmaBaselineState {
    pub sample_count: u64,
    pub ema_mean: f64,
    pub ema_variance: f64,
    pub last_update: u64,
}

impl EmaBaselineState {
    pub fn update(&mut self, sample: f64, alpha: f64, now: u64) {
        let alpha = alpha.clamp(f64::MIN_POSITIVE, 1.0);
        if self.sample_count == 0 {
            self.ema_mean = sample;
            self.ema_variance = 0.0;
        } else {
            let prev_mean = self.ema_mean;
            self.ema_mean = prev_mean + alpha * (sample - prev_mean);
            // Incremental EWMA variance, following West (1979) / Welford.
            let diff = sample - prev_mean;
            self.ema_variance = (1.0 - alpha) * (self.ema_variance + alpha * diff * diff);
        }
        self.sample_count = self.sample_count.saturating_add(1);
        self.last_update = now;
    }

    pub fn stddev(&self) -> f64 {
        self.ema_variance.max(0.0).sqrt()
    }
}

Two things to read out of the update body:

• First sample seeds the mean. When sample_count == 0 the very first call sets ema_mean = sample and ema_variance = 0.0. The arithmetic prev_mean + alpha * (sample - prev_mean) (which is just the standard EMA update) does not run on sample one; it runs from sample two onward.
• EWMA variance, not sample variance. The variance update (1 - alpha) * (ema_variance + alpha * diff * diff) is the West/Welford incremental EWMA form. stddev() is sqrt(max(ema_variance, 0.0)); there is no Bessel correction and no separate sample-variance path.

Robust z-score with Poisson floor

The guard does not call EmaBaselineState::z_score directly; it uses a variant that clamps stddev away from zero (behavioral_profile.rs:337-348):

fn robust_z_score(state: &EmaBaselineState, sample: f64) -> Option<f64> {
    if state.sample_count < 2 {
        return None;
    }
    let measured = state.stddev();
    let floor = state.ema_mean.max(1.0).sqrt();
    let effective = measured.max(floor);
    if effective <= f64::EPSILON {
        return None;
    }
    Some((sample - state.ema_mean) / effective)
}

The early return on sample_count < 2 is the documented reason the first sample never updates a usable baseline: observe_sample calls robust_z_score before EmaBaselineState::update, so the very first observation always returns z_score = None and anomaly = false. The Poisson floor sqrt(max(mean, 1)) means a 50x spike over a steady 10/window baseline still flags even when EWMA variance is numerically zero.

Window-start quantization

The current-window calculation (behavioral_profile.rs:294-297) is a plain integer-division quantizer:

fn current_window_start(&self, now: u64) -> u64 {
    let window = self.config.window_secs.max(1);
    (now / window) * window
}

now is in unix seconds (behavioral_profile.rs:322-326), and the window_secs.max(1) guards a misconfigured zero. Two consequences:

• Calls within the same window-start bucket fold into one sample. observe_sample short-circuits when last_window_start == window_start (behavioral_profile.rs:246-257) and returns the cached outcome without bumping sample_count.
• There are no sub-second timestamps. The clock source is SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()) (behavioral_profile.rs:323-326), so anything finer than 1 s is discarded before the divisor sees it.

Metrics

From BehavioralMetric (behavioral_profile.rs:57-79):

• CallRate :: "call_rate", receipts per window.
• DenyRate :: "deny_rate", denies per window.
• UniqueTools :: "unique_tools", distinct tool names per window.
• AvgParameterEntropy :: "avg_parameter_entropy", Shannon entropy of parameters.

On the synchronous Guard::evaluate path (behavioral_profile.rs:355-365), only CallRate is sampled: the count is receipts.len() as f64 from sample_for_window (behavioral_profile.rs:299-305). The other three metrics are reachable through observe_sample for callers that want to feed values out-of-band (an offline batch or a dashboard).

BehavioralAnomalyScore

The dashboard-visible struct paired with EmaBaselineState (operator_report/behavioral_analysis.rs:215-236). The synchronous guard never fills it: evaluate discards its observation outcome and returns a bare GuardDecision::allow(). The struct is instead assembled by a caller that invokes observe_sample directly — the operator-report path or a dashboard — and reads the ObservationOutcome it returns:

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct BehavioralAnomalyScore {
    pub agent_id: String,
    pub baseline: EmaBaselineState,
    pub current_sample: f64,
    pub z_score: Option<f64>,
    pub sigma_threshold: f64,
    pub anomaly: bool,
    pub generated_at: u64,
}

The matching per-call return type is ObservationOutcome (behavioral_profile.rs:309-320), which pairs the same z_score, anomaly, baseline, and sample fields.

Algorithm

1. window_start = (now / window_secs) * window_secs (behavioral_profile.rs:294-297).
2. Read receipts in [window_start, window_end - 1] where window_end = window_start + window_secs.max(1) via the ReceiptFeedSource (behavioral_profile.rs:299-305). The upper bound is exclusive after thesaturating_sub(1).
3. Sample = receipts.len() as f64.
4. Compute robust_z_score(&state, sample) against the pre-update baseline. If sample_count >= baseline_min_windows and |z| > sigma_threshold, mark anomaly = true (behavioral_profile.rs:259-263).
5. Update the baseline with the new sample (one update per window-start; repeated calls in the same bucket short-circuit at behavioral_profile.rs:246-257).
6. Return GuardDecision::allow(). The observation outcome is discarded — let _ = self.observe_sample(...)? — and allow() carries an empty evidence array, so the computed z-score, anomaly flag, and baseline never reach this guard's GuardDecision and never land in the signed receipt through the synchronous evaluate path. The one durable effect of the call is folding the new sample into the per-agent EMA baseline for the next window.

Failure modes

• Mutex poisoning :: Err(KernelError::Internal("baseline lock poisoned")) (behavioral_profile.rs:240).
• Receipt feed error :: propagated through ? (behavioral_profile.rs:359). The kernel reads it as deny even though the normal verdict is advisory.
• Cold baseline (sample count below baseline_min_windows) :: never flags. The first observation also returns z_score = None from the sample_count < 2 early return.

Storage

Baselines live in memory keyed by (agent_id, BehavioralMetric) behind a single Mutex<HashMap<...>> (behavioral_profile.rs:199). The receipt feed is pluggable through ReceiptFeedSource (behavioral_profile.rs:84-95); InMemoryReceiptFeed (behavioral_profile.rs:104-158) ships with the crate for tests and lightweight deployments. Production wiring backs the trait with ReceiptStore::query_receipts from chio-store-sqlite.

Composition

use std::sync::Arc;
use chio_guards::{
    DataFlowGuard, DataFlowConfig,
    BehavioralSequenceGuard, SequencePolicy,
    BehavioralProfileGuard, BehavioralProfileConfig,
    InMemoryReceiptFeed,
};
use chio_http_session::SessionJournal;

let journal = Arc::new(SessionJournal::new("sess-1".to_string()));
let feed = InMemoryReceiptFeed::new();

let mut pipeline = chio_guards::GuardPipeline::new();

pipeline.add(Box::new(DataFlowGuard::new(
    journal.clone(),
    DataFlowConfig {
        max_bytes_read: Some(50 * 1024 * 1024),
        max_bytes_written: Some(10 * 1024 * 1024),
        max_bytes_total: None,
    },
)));

let mut policy = SequencePolicy::default();
policy.required_first_tool = Some("init".to_string());
pipeline.add(Box::new(BehavioralSequenceGuard::new(journal.clone(), policy)));

pipeline.add(Box::new(BehavioralProfileGuard::with_config(
    Box::new(feed),
    BehavioralProfileConfig::default(),
)));

Next Steps

• Rate Limit Guards :: complementary token-bucket throttles that do not require the journal.
• Fail-Closed Behavior :: how journal errors translate to verdicts.
• Query Audit Receipts :: the receipt store that backs BehavioralProfileGuard in production.