daemon/grpc_local_impl/
signal.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Native signal-health queries used by the local CLI.
3
4use std::collections::{HashMap, HashSet};
5
6use objects::{
7    object::{RiskSignal, RiskSignalBlob, State, StateAttachmentBody},
8    store::ObjectStore,
9};
10use repo::{Repository, StateAttachmentKind};
11
12const DEFAULT_HEALTH_WINDOW: usize = 200;
13const MAX_HEALTH_WINDOW: u32 = 5_000;
14
15#[derive(Debug, Clone, PartialEq)]
16pub struct SignalHealthEntry {
17    pub module_id: String,
18    pub fire_rate: f32,
19    pub warn: bool,
20}
21
22#[derive(Debug, Clone, PartialEq)]
23pub struct SignalHealthReport {
24    pub entries: Vec<SignalHealthEntry>,
25    pub window_states: u32,
26}
27
28/// Report per-module signal fire rates over the repository's recent
29/// first-parent history. This is deliberately a native query: signal health
30/// is local CLI behavior, not part of the shared hosted contract.
31pub fn get_repo_signal_health(
32    repo: &Repository,
33    requested_window: u32,
34) -> objects::error::Result<SignalHealthReport> {
35    let window = if requested_window == 0 {
36        DEFAULT_HEALTH_WINDOW
37    } else {
38        requested_window.min(MAX_HEALTH_WINDOW) as usize
39    };
40    let states = walk_recent_states(repo, window)?;
41    let visited = states.len() as u32;
42    let mut per_module: HashMap<String, u32> = HashMap::new();
43    for state in &states {
44        let mut seen_modules = HashSet::new();
45        for signal in load_signals(repo, state)? {
46            let module = signal.producer.module;
47            if seen_modules.insert(module.clone()) {
48                *per_module.entry(module).or_default() += 1;
49            }
50        }
51    }
52    let mut entries: Vec<_> = per_module
53        .into_iter()
54        .map(|(module_id, hit_count)| {
55            let fire_rate = if visited == 0 {
56                0.0
57            } else {
58                hit_count as f32 / visited as f32
59            };
60            SignalHealthEntry {
61                module_id,
62                fire_rate,
63                warn: fire_rate > 0.5,
64            }
65        })
66        .collect();
67    entries.sort_by(|a, b| a.module_id.cmp(&b.module_id));
68    Ok(SignalHealthReport {
69        entries,
70        window_states: visited,
71    })
72}
73
74fn load_signals(repo: &Repository, state: &State) -> objects::error::Result<Vec<RiskSignal>> {
75    let Some(attachment) =
76        repo.latest_state_attachment(&state.state_id, StateAttachmentKind::RiskSignals)?
77    else {
78        return Ok(Vec::new());
79    };
80    let StateAttachmentBody::RiskSignals(hash) = attachment.body else {
81        unreachable!()
82    };
83    let blob = repo.store().get_blob(&hash)?.ok_or_else(|| {
84        objects::error::HeddleError::InvalidObject(format!(
85            "risk-signals blob {hash} referenced by state {} is missing",
86            state.state_id
87        ))
88    })?;
89    RiskSignalBlob::decode(blob.content())
90        .map(|parsed| parsed.signals)
91        .map_err(|error| objects::error::HeddleError::InvalidObject(error.to_string()))
92}
93
94fn walk_recent_states(repo: &Repository, window: usize) -> objects::error::Result<Vec<State>> {
95    let mut states = Vec::new();
96    let mut cursor = repo.head()?;
97    while let Some(id) = cursor {
98        if states.len() >= window {
99            break;
100        }
101        let Some(state) = repo.store().get_state(&id)? else {
102            break;
103        };
104        cursor = state.parents.first().copied();
105        states.push(state);
106    }
107    Ok(states)
108}
daemon/grpc_local_impl/signal.rs

daemon/grpc_local_impl/
signal.rs
