mati_core/store/
observability.rs

1//! Read-only aggregation of policy enforcement telemetry.
2//!
3//! These are pure transforms: callers scan the raw records themselves (the CLI
4//! through `StoreProxy`, the MCP `mem_query` handler through `&Store`) and hand
5//! the record sets here. Keeping the aggregation math in one place means the
6//! `mati policy` CLI and the `mem_query` telemetry modes cannot drift — the same
7//! records always produce the same report, whichever door reached them.
8//!
9//! Nothing here computes a verdict or a recommendation. It exposes the local
10//! aggregates the store already holds, with their provenance (`sources`,
11//! `last_fired_at`, retention bounds) intact. Reporting on top of these numbers
12//! lives outside this repo.
13
14use std::collections::{BTreeMap, BTreeSet};
15
16use serde::Serialize;
17
18use crate::store::enforcement::{EnforcementEventScan, EnforcementEventType, SubjectKind};
19use crate::store::record::{PolicyRecord, PolicyStage, Record, RecordLifecycle};
20use crate::store::session::{
21    DailyAgg, PolicyShadowAgg, ShadowObservationAgg, MAX_SHADOW_OBSERVATIONS,
22};
23
24/// Enforcement events are retained for a bounded window; activity claims older
25/// than this cannot be made.
26pub const POLICY_ACTIVITY_RETENTION_DAYS: u64 = 365;
27/// Default look-back window for `mati policy activity` and the `policy_activity`
28/// query mode.
29pub const POLICY_ACTIVITY_DEFAULT_DAYS: u64 = 30;
30/// A newly enabled policy is given this long before `mati doctor` flags it as
31/// inactive — a quiet first week is not yet evidence the rule is dead.
32pub const POLICY_ACTIVITY_GRACE_DAYS: u64 = 7;
33
34/// Whether a policy has fired in the window, or why the question can't be
35/// answered.
36#[derive(Debug, Clone, Serialize)]
37#[serde(rename_all = "snake_case")]
38pub enum ActivityState {
39    Fired,
40    NoActivity,
41    NotMeasurable,
42}
43
44/// Per-policy activity over the requested window.
45#[derive(Debug, Clone, Serialize)]
46pub struct PolicyActivity {
47    pub policy: String,
48    pub state: ActivityState,
49    pub window_days: u64,
50    pub window_start: u64,
51    pub count: u64,
52    pub last_fired_at: Option<u64>,
53    pub sources: Vec<String>,
54}
55
56/// The full activity report for all active, measurable policies.
57#[derive(Debug, Clone, Serialize)]
58pub struct ActivityReport {
59    pub window_days: u64,
60    pub window_start: u64,
61    pub window_end: u64,
62    pub retention_days: u64,
63    pub retention_limited: bool,
64    pub retention_note: Option<String>,
65    pub policies: Vec<PolicyActivity>,
66}
67
68/// Start of the look-back window, in epoch seconds. Saturates so a huge `days`
69/// can never underflow past the epoch.
70pub fn window_start_secs(now_secs: u64, days: u64) -> u64 {
71    now_secs.saturating_sub(days.saturating_mul(86_400))
72}
73
74/// Accept either a bare slug (`my-rule`) or a full key (`policy:my-rule`).
75fn normalize_policy_key(slug: &str) -> String {
76    if slug.starts_with("policy:") {
77        slug.to_string()
78    } else {
79        format!("policy:{slug}")
80    }
81}
82
83/// Merge the bounded daily shadow aggregates into a per-policy view, optionally
84/// filtered to one policy. `shadow_records` is the `analytics:policy_shadow_*`
85/// scan; observations are kept sorted and capped at `MAX_SHADOW_OBSERVATIONS`.
86pub fn assemble_shadow_observations(
87    shadow_records: &[Record],
88    slug: Option<&str>,
89) -> BTreeMap<String, PolicyShadowAgg> {
90    let mut observations: BTreeMap<String, PolicyShadowAgg> = BTreeMap::new();
91    for record in shadow_records {
92        let agg = record
93            .payload_as::<ShadowObservationAgg>()
94            .unwrap_or_default();
95        for (key, mut policy) in agg.policies {
96            let entry = observations.entry(key).or_default();
97            entry.count += policy.count;
98            entry.observations.append(&mut policy.observations);
99            entry
100                .observations
101                .sort_by_key(|observation| observation.timestamp);
102            if entry.observations.len() > MAX_SHADOW_OBSERVATIONS {
103                let drop_count = entry.observations.len() - MAX_SHADOW_OBSERVATIONS;
104                entry.observations.drain(..drop_count);
105            }
106        }
107    }
108    if let Some(slug) = slug {
109        let key = normalize_policy_key(slug);
110        observations.retain(|policy_key, _| policy_key == &key);
111    }
112    observations
113}
114
115/// Build the activity report from pre-scanned record sets.
116///
117/// `policy_records` is the `policy:*` scan, `enforcement` a time-bounded
118/// enforcement scan over `[window_start, now]`, and the shadow/steer slices are
119/// the `analytics:policy_shadow_*` / `analytics:policy_steer_*` scans.
120///
121/// The shadow and steer records are window-filtered here (`updated_at <
122/// window_start`), but the enforcement events are counted VERBATIM — windowing
123/// them is the caller's job, done by scanning only `[window_start_secs(now_secs,
124/// days), now]`. Hand this a wider enforcement scan and `count`, `state`, and
125/// `last_fired_at` inflate past the window, not just the retention note.
126pub fn assemble_activity_report(
127    now_secs: u64,
128    days: u64,
129    policy_records: &[Record],
130    enforcement: &EnforcementEventScan,
131    shadow_records: &[Record],
132    steer_records: &[Record],
133    slug: Option<&str>,
134) -> ActivityReport {
135    let window_start = window_start_secs(now_secs, days);
136
137    let mut policies = BTreeMap::<String, PolicyRecord>::new();
138    for record in policy_records {
139        if matches!(record.lifecycle, RecordLifecycle::Active) {
140            if let Some(policy) = record.payload_as::<PolicyRecord>() {
141                if !matches!(policy.stage, PolicyStage::Off) {
142                    policies.insert(record.key.clone(), policy);
143                }
144            }
145        }
146    }
147
148    let mut counts = BTreeMap::<String, u64>::new();
149    let mut last = BTreeMap::<String, u64>::new();
150    let mut sources = BTreeMap::<String, BTreeSet<String>>::new();
151    for event in &enforcement.events {
152        if !matches!(event.subject_kind, SubjectKind::Control)
153            || !event.subject_key.starts_with("policy:")
154            || !matches!(
155                event.event_type,
156                EnforcementEventType::Deny | EnforcementEventType::AllowAfterReceipt
157            )
158        {
159            continue;
160        }
161        let key = event.subject_key.clone();
162        *counts.entry(key.clone()).or_default() += 1;
163        last.entry(key.clone())
164            .and_modify(|value| *value = (*value).max(event.recorded_at_ms / 1000))
165            .or_insert(event.recorded_at_ms / 1000);
166        sources.entry(key).or_default().insert("enforcement".into());
167    }
168
169    for record in shadow_records {
170        if record.updated_at < window_start {
171            continue;
172        }
173        let agg = record
174            .payload_as::<ShadowObservationAgg>()
175            .unwrap_or_default();
176        for (key, policy) in agg.policies {
177            if policy.count == 0 {
178                continue;
179            }
180            *counts.entry(key.clone()).or_default() += policy.count;
181            if let Some(timestamp) = policy.observations.iter().map(|o| o.timestamp).max() {
182                last.entry(key.clone())
183                    .and_modify(|value| *value = (*value).max(timestamp))
184                    .or_insert(timestamp);
185            }
186            sources.entry(key).or_default().insert("shadow".into());
187        }
188    }
189
190    for record in steer_records {
191        if record.updated_at < window_start {
192            continue;
193        }
194        let Some(agg) = record.payload_as::<DailyAgg>() else {
195            continue;
196        };
197        for (key, count) in agg.key_counts {
198            if count == 0 {
199                continue;
200            }
201            *counts.entry(key.clone()).or_default() += count;
202            last.entry(key.clone())
203                .and_modify(|value| *value = (*value).max(record.updated_at))
204                .or_insert(record.updated_at);
205            sources.entry(key).or_default().insert("steer".into());
206        }
207    }
208
209    let retention_limited = enforcement
210        .oldest_recorded_at_ms
211        .is_some_and(|oldest| window_start.saturating_mul(1000) < oldest);
212    let mut report = ActivityReport {
213        window_days: days,
214        window_start,
215        window_end: now_secs,
216        retention_days: POLICY_ACTIVITY_RETENTION_DAYS,
217        retention_limited,
218        retention_note: retention_limited.then(|| format!(
219            "requested window predates the oldest retained enforcement event; history is bounded to {} days",
220            POLICY_ACTIVITY_RETENTION_DAYS
221        )),
222        policies: Vec::new(),
223    };
224    for (key, policy) in policies {
225        // The matcher understands file_read, but no adapter routes that action
226        // through the policy gate. Unknown tool names have the same absence of
227        // a trace. Keep both as not-measurable rather than calling them dead.
228        let measurable = match policy.trigger.tool.as_deref() {
229            None | Some("db_client") | Some("path") => true,
230            Some(_) => false,
231        };
232        let count = counts.get(&key).copied().unwrap_or(0);
233        let state = if !measurable {
234            ActivityState::NotMeasurable
235        } else if count > 0 {
236            ActivityState::Fired
237        } else {
238            ActivityState::NoActivity
239        };
240        report.policies.push(PolicyActivity {
241            policy: key.clone(),
242            state,
243            window_days: days,
244            window_start,
245            count,
246            last_fired_at: last.get(&key).copied(),
247            sources: sources
248                .remove(&key)
249                .unwrap_or_default()
250                .into_iter()
251                .collect(),
252        });
253    }
254    report.policies.sort_by(|a, b| a.policy.cmp(&b.policy));
255    if let Some(slug) = slug {
256        let key = normalize_policy_key(slug);
257        report.policies.retain(|policy| policy.policy == key);
258    }
259    report
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265    use crate::store::enforcement::EnforcementEvent;
266    use crate::store::record::{
267        PolicyFreshness, PolicyMode, PolicyRequires, PolicyTrigger, Priority, TombstoneReason,
268    };
269
270    /// Build an `analytics:*`-style record with a payload and `updated_at`,
271    /// reusing the production analytics-record constructor for field fidelity.
272    fn agg_record(key: &str, updated_at: u64, payload: serde_json::Value) -> Record {
273        let mut r = crate::store::session::analytics_record(key, String::new());
274        r.updated_at = updated_at;
275        r.payload = Some(payload);
276        r
277    }
278
279    fn policy_record(key: &str, tool: Option<&str>, stage: PolicyStage) -> Record {
280        let policy = PolicyRecord {
281            name: "n".into(),
282            rule: "r".into(),
283            reason: "why".into(),
284            scope: "s".into(),
285            mode: PolicyMode::Block,
286            trigger: PolicyTrigger {
287                tool: tool.map(String::from),
288                ..Default::default()
289            },
290            requires: PolicyRequires {
291                key: "gotcha:x".into(),
292                via: vec![],
293                freshness: PolicyFreshness {
294                    ttl_secs: 900,
295                    fingerprint: false,
296                },
297            },
298            stage,
299            severity: Priority::High,
300            created_by: "test".into(),
301        };
302        agg_record(key, 0, serde_json::to_value(policy).unwrap())
303    }
304
305    fn deny_event(subject_key: &str, recorded_at_ms: u64, seq_no: u64) -> EnforcementEvent {
306        EnforcementEvent {
307            event_id: format!("evt-{seq_no}"),
308            schema_version: 1,
309            seq_no,
310            recorded_at_ms,
311            event_type: EnforcementEventType::Deny,
312            event_hash: String::new(),
313            prev_hash: String::new(),
314            installation_id: "test".into(),
315            actor_local: None,
316            agent_type: "codex".into(),
317            subject_kind: SubjectKind::Control,
318            subject_key: subject_key.into(),
319            canonical_subject_hash: None,
320            receipt_id: None,
321            decision_reason_code: "policy_deny".into(),
322            decision_basis_hash: None,
323            agent_session: None,
324            agent_id: None,
325            parent_agent_id: None,
326        }
327    }
328
329    fn scan_of(events: Vec<EnforcementEvent>, oldest: Option<u64>) -> EnforcementEventScan {
330        EnforcementEventScan {
331            events,
332            oldest_recorded_at_ms: oldest,
333            scanned_keys: 0,
334        }
335    }
336
337    fn shadow_record(key: &str, updated_at: u64, policy_key: &str, count: u64) -> Record {
338        let mut policies = BTreeMap::new();
339        policies.insert(
340            policy_key.to_string(),
341            PolicyShadowAgg {
342                count,
343                observations: vec![],
344            },
345        );
346        agg_record(
347            key,
348            updated_at,
349            serde_json::to_value(ShadowObservationAgg { policies }).unwrap(),
350        )
351    }
352
353    #[test]
354    fn window_start_saturates_at_epoch() {
355        assert_eq!(window_start_secs(100, 0), 100);
356        assert_eq!(window_start_secs(100, 1), 100u64.saturating_sub(86_400));
357        assert_eq!(window_start_secs(1_000_000, 1), 1_000_000 - 86_400);
358    }
359
360    #[test]
361    fn normalize_accepts_bare_and_full_keys() {
362        assert_eq!(normalize_policy_key("my-rule"), "policy:my-rule");
363        assert_eq!(normalize_policy_key("policy:my-rule"), "policy:my-rule");
364    }
365
366    #[test]
367    fn empty_inputs_produce_empty_report() {
368        let scan = EnforcementEventScan {
369            events: Vec::new(),
370            oldest_recorded_at_ms: None,
371            scanned_keys: 0,
372        };
373        let report = assemble_activity_report(1_000_000, 30, &[], &scan, &[], &[], None);
374        assert!(report.policies.is_empty());
375        assert!(!report.retention_limited);
376        assert_eq!(report.window_days, 30);
377        assert_eq!(report.retention_days, POLICY_ACTIVITY_RETENTION_DAYS);
378    }
379
380    #[test]
381    fn observations_of_empty_scan_are_empty() {
382        assert!(assemble_shadow_observations(&[], None).is_empty());
383        assert!(assemble_shadow_observations(&[], Some("anything")).is_empty());
384    }
385
386    #[test]
387    fn enforcement_counted_regardless_of_window_but_shadow_is_gated() {
388        let now = 1_000_000_000u64;
389        let days = 1; // window_start = now - 86_400
390        let policy = policy_record("policy:p", Some("db_client"), PolicyStage::Enforce);
391        // Enforcement event recorded long before the window: assemble counts it
392        // verbatim — windowing is the caller's scan responsibility.
393        let old_ms = (now - 10 * 86_400) * 1000;
394        let scan = scan_of(vec![deny_event("policy:p", old_ms, 1)], Some(old_ms));
395        // Shadow record predating the window: assemble MUST drop it.
396        let stale = shadow_record(
397            "analytics:policy_shadow_old",
398            now - 10 * 86_400,
399            "policy:p",
400            5,
401        );
402        let report = assemble_activity_report(now, days, &[policy], &scan, &[stale], &[], None);
403        let p = &report.policies[0];
404        assert_eq!(
405            p.count, 1,
406            "enforcement counted despite predating the window"
407        );
408        assert!(matches!(p.state, ActivityState::Fired));
409        assert_eq!(
410            p.sources,
411            vec!["enforcement".to_string()],
412            "the stale shadow record must not add a 'shadow' source"
413        );
414    }
415
416    #[test]
417    fn measurable_state_is_a_pure_function_of_trigger_tool() {
418        let now = 2_000_000_000u64;
419        let scan = scan_of(
420            vec![deny_event("policy:bash", now * 1000, 1)],
421            Some(now * 1000),
422        );
423        let policies = vec![
424            policy_record("policy:bash", Some("bash"), PolicyStage::Enforce),
425            policy_record("policy:db", Some("db_client"), PolicyStage::Enforce),
426            policy_record("policy:none", None, PolicyStage::Enforce),
427        ];
428        let report = assemble_activity_report(now, 30, &policies, &scan, &[], &[], None);
429        let state = |k: &str| {
430            report
431                .policies
432                .iter()
433                .find(|p| p.policy == k)
434                .unwrap()
435                .state
436                .clone()
437        };
438        // Even with a matching Deny, an unmeasurable tool is NotMeasurable —
439        // never Fired. This is the one derived field, and it stays mechanical.
440        assert!(matches!(state("policy:bash"), ActivityState::NotMeasurable));
441        assert!(matches!(state("policy:db"), ActivityState::NoActivity));
442        assert!(matches!(state("policy:none"), ActivityState::NoActivity));
443    }
444
445    #[test]
446    fn stale_shadow_shows_in_observations_but_not_in_activity() {
447        let now = 3_000_000_000u64;
448        let stale = shadow_record(
449            "analytics:policy_shadow_old",
450            now - 60 * 86_400,
451            "policy:p",
452            3,
453        );
454        // observations: no window filter — the count survives.
455        let obs = assemble_shadow_observations(std::slice::from_ref(&stale), None);
456        assert_eq!(obs["policy:p"].count, 3);
457        // activity (30d): the record predates the window — contributes nothing.
458        let report = assemble_activity_report(
459            now,
460            30,
461            &[policy_record(
462                "policy:p",
463                Some("path"),
464                PolicyStage::Enforce,
465            )],
466            &scan_of(vec![], None),
467            &[stale],
468            &[],
469            None,
470        );
471        assert_eq!(report.policies[0].count, 0);
472        assert!(report.policies[0].sources.is_empty());
473    }
474
475    #[test]
476    fn off_and_tombstoned_policies_excluded_and_retention_flagged() {
477        let now = 4_000_000_000u64;
478        let off = policy_record("policy:off", None, PolicyStage::Off);
479        let mut dead = policy_record("policy:dead", None, PolicyStage::Enforce);
480        dead.lifecycle = RecordLifecycle::Tombstoned {
481            reason: TombstoneReason::ManualDeletion,
482            at: now,
483        };
484        let live = policy_record("policy:live", None, PolicyStage::Enforce);
485        // A 400-day window starts before the oldest retained event (10 days
486        // old) → the requested history predates retention → retention_limited.
487        let old_ms = (now - 10 * 86_400) * 1000;
488        let scan = scan_of(vec![deny_event("policy:live", old_ms, 1)], Some(old_ms));
489        let report = assemble_activity_report(now, 400, &[off, dead, live], &scan, &[], &[], None);
490        assert_eq!(report.policies.len(), 1);
491        assert_eq!(report.policies[0].policy, "policy:live");
492        assert!(report.retention_limited);
493        assert!(report.retention_note.is_some());
494    }
495}
mati_core/store/observability.rs

mati_core/store/
observability.rs
