use std::collections::HashMap;
use std::sync::Mutex as StdMutex;
use std::time::Instant;
use chrono::Utc;
use sha2::{Digest, Sha256};
use thiserror::Error;
use tokio::sync::Mutex;
use tracing::{debug, warn};
use super::cache::{ActivityCache, ActivityState, ActivityVerdict, CheckMetrics, CostTally};
#[derive(Debug, Default, Clone)]
struct SummarySlot {
last_summary: Option<String>,
summarizing: bool,
}
#[derive(Debug, Error)]
pub enum ActivityError {
#[error("LLM classification error: {0}")]
Llm(String),
#[error("serialization error: {0}")]
Serialization(String),
#[error("OPENROUTER_API_KEY is not configured")]
MissingApiKey,
}
pub trait LlmClassifier: Send + Sync {
fn classify(
&self,
pane_text: &str,
) -> impl Future<Output = Result<(ActivityVerdict, u32, u32), ActivityError>> + Send;
}
#[derive(Debug)]
pub struct ActivityCheckResult {
pub verdict: ActivityVerdict,
pub cost: CheckMetrics,
pub cache_hit: bool,
pub tally: CostTally,
}
pub struct ActivityMonitor<C: LlmClassifier> {
cache: Mutex<HashMap<String, ActivityCache>>,
summaries: StdMutex<HashMap<String, SummarySlot>>,
llm: C,
model: String,
}
impl<C: LlmClassifier> std::fmt::Debug for ActivityMonitor<C> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ActivityMonitor")
.field("model", &self.model)
.finish_non_exhaustive()
}
}
impl<C: LlmClassifier> ActivityMonitor<C> {
pub fn new(llm: C, model: impl Into<String>) -> Self {
Self {
cache: Mutex::new(HashMap::new()),
summaries: StdMutex::new(HashMap::new()),
llm,
model: model.into(),
}
}
fn lock_summaries(&self) -> std::sync::MutexGuard<'_, HashMap<String, SummarySlot>> {
self.summaries.lock().unwrap_or_else(|e| {
warn!("activity monitor: summaries mutex poisoned, recovering");
e.into_inner()
})
}
pub fn cached_summary(&self, session_id: &str) -> Option<String> {
let slots = self.lock_summaries();
slots.get(session_id).and_then(|s| s.last_summary.clone())
}
pub fn is_summarizing(&self, session_id: &str) -> bool {
let slots = self.lock_summaries();
slots.get(session_id).is_some_and(|s| s.summarizing)
}
fn set_summarizing(&self, session_id: &str, flag: bool) {
let mut slots = self.lock_summaries();
slots.entry(session_id.to_owned()).or_default().summarizing = flag;
}
fn store_summary(&self, session_id: &str, summary: String) {
let mut slots = self.lock_summaries();
slots.entry(session_id.to_owned()).or_default().last_summary = Some(summary);
}
pub async fn check(
&self,
session_id: &str,
pane_text: &str,
) -> Result<ActivityCheckResult, ActivityError> {
let pane_tail = last_n_lines(pane_text, 60);
let hash = sha256_hex(pane_tail.as_bytes());
let start = Instant::now();
let mut caches = self.cache.lock().await;
let entry = caches
.entry(session_id.to_owned())
.or_insert_with(|| ActivityCache::new(&self.model));
if entry.check_unchanged(&hash) {
let verdict = entry
.last_verdict()
.cloned()
.unwrap_or_else(|| ActivityVerdict {
state: ActivityState::Unknown,
summary: "no prior verdict in cache".into(),
confidence: 0.0,
});
let metrics = CheckMetrics {
session_id: session_id.to_owned(),
at: Utc::now(),
model: self.model.clone(),
input_tokens: 0,
output_tokens: 0,
latency_ms: start.elapsed().as_millis() as u64,
cache_hit: true,
verdict_state: verdict.state.clone(),
};
let tally = entry.tally().clone();
entry.update_cache_hit(&hash, metrics.clone());
debug!(session = %session_id, "activity check: cache hit");
return Ok(ActivityCheckResult {
verdict,
cost: metrics,
cache_hit: true,
tally,
});
}
drop(caches);
let _summarizing = SummarizingGuard::new(self, session_id);
let classify_result = self.llm.classify(&pane_tail).await;
let (verdict, input_tokens, output_tokens) = match classify_result {
Ok(r) => r,
Err(ActivityError::MissingApiKey) => {
warn!("activity monitor: OPENROUTER_API_KEY not configured; returning Unknown");
(
ActivityVerdict {
state: ActivityState::Unknown,
summary: "OPENROUTER_API_KEY not configured".into(),
confidence: 0.0,
},
0,
0,
)
}
Err(e) => return Err(e),
};
if verdict.state != ActivityState::Unknown {
self.store_summary(session_id, verdict.summary.clone());
}
let latency_ms = start.elapsed().as_millis() as u64;
let metrics = CheckMetrics {
session_id: session_id.to_owned(),
at: Utc::now(),
model: self.model.clone(),
input_tokens,
output_tokens,
latency_ms,
cache_hit: false,
verdict_state: verdict.state.clone(),
};
let mut caches = self.cache.lock().await;
let entry = caches
.entry(session_id.to_owned())
.or_insert_with(|| ActivityCache::new(&self.model));
entry.update_llm_hit(&hash, verdict.clone(), metrics.clone());
let tally = entry.tally().clone();
debug!(session = %session_id, state = ?verdict.state, "activity check: LLM verdict");
Ok(ActivityCheckResult {
verdict,
cost: metrics,
cache_hit: false,
tally,
})
}
}
struct SummarizingGuard<'a, C: LlmClassifier> {
monitor: &'a ActivityMonitor<C>,
session_id: &'a str,
}
impl<'a, C: LlmClassifier> SummarizingGuard<'a, C> {
fn new(monitor: &'a ActivityMonitor<C>, session_id: &'a str) -> Self {
monitor.set_summarizing(session_id, true);
Self {
monitor,
session_id,
}
}
}
impl<C: LlmClassifier> Drop for SummarizingGuard<'_, C> {
fn drop(&mut self) {
self.monitor.set_summarizing(self.session_id, false);
}
}
fn sha256_hex(bytes: &[u8]) -> String {
let mut h = Sha256::new();
h.update(bytes);
let digest = h.finalize();
let mut hex = String::with_capacity(digest.len() * 2);
for b in digest {
use std::fmt::Write as _;
let _ = write!(hex, "{b:02x}");
}
hex
}
fn last_n_lines(text: &str, n: usize) -> String {
let lines: Vec<&str> = text.lines().collect();
let start = lines.len().saturating_sub(n);
lines[start..].join("\n")
}
use std::future::Future;
#[cfg(test)]
mod tests {
use super::*;
struct MockClassifier {
verdict: ActivityVerdict,
call_count: Mutex<u32>,
}
impl MockClassifier {
fn new(state: ActivityState) -> Self {
Self {
verdict: ActivityVerdict {
state,
summary: "mock".into(),
confidence: 1.0,
},
call_count: Mutex::new(0),
}
}
async fn calls(&self) -> u32 {
*self.call_count.lock().await
}
}
impl LlmClassifier for MockClassifier {
async fn classify(
&self,
_pane_text: &str,
) -> Result<(ActivityVerdict, u32, u32), ActivityError> {
*self.call_count.lock().await += 1;
Ok((self.verdict.clone(), 50, 10))
}
}
#[tokio::test]
async fn monitor_cache_miss_calls_llm() {
let classifier = MockClassifier::new(ActivityState::Working);
let monitor = ActivityMonitor::new(classifier, "test-model");
let result = monitor.check("s1", "some pane content").await.unwrap();
assert_eq!(result.verdict.state, ActivityState::Working);
assert!(!result.cache_hit);
assert_eq!(monitor.llm.calls().await, 1);
}
#[tokio::test]
async fn monitor_cache_hit_skips_llm() {
let classifier = MockClassifier::new(ActivityState::Idle);
let monitor = ActivityMonitor::new(classifier, "test-model");
let pane = "unchanged content";
let r1 = monitor.check("s1", pane).await.unwrap();
assert!(!r1.cache_hit);
let r2 = monitor.check("s1", pane).await.unwrap();
assert!(r2.cache_hit);
assert_eq!(monitor.llm.calls().await, 1);
}
#[tokio::test]
async fn cached_summary_none_without_check() {
let classifier = MockClassifier::new(ActivityState::Working);
let monitor = ActivityMonitor::new(classifier, "test-model");
assert_eq!(monitor.cached_summary("s1"), None);
assert!(!monitor.is_summarizing("s1"));
}
#[tokio::test]
async fn cached_summary_reflects_llm_verdict() {
let classifier = MockClassifier::new(ActivityState::Working);
let monitor = ActivityMonitor::new(classifier, "test-model");
monitor.check("s1", "pane").await.unwrap();
assert_eq!(monitor.cached_summary("s1").as_deref(), Some("mock"));
}
#[tokio::test]
async fn cached_summary_skips_unknown_verdict() {
let classifier = MockClassifier::new(ActivityState::Unknown);
let monitor = ActivityMonitor::new(classifier, "test-model");
monitor.check("s1", "pane").await.unwrap();
assert_eq!(monitor.cached_summary("s1"), None);
}
#[tokio::test]
async fn summarizing_flag_is_false_when_idle() {
let classifier = MockClassifier::new(ActivityState::Working);
let monitor = ActivityMonitor::new(classifier, "test-model");
assert!(!monitor.is_summarizing("s1"));
monitor.check("s1", "pane").await.unwrap();
assert!(!monitor.is_summarizing("s1"));
}
#[tokio::test]
async fn summarizing_true_during_inference() {
use std::sync::Arc;
use tokio::sync::Notify;
struct BlockingClassifier {
entered: Arc<Notify>,
release: Arc<Notify>,
}
impl LlmClassifier for BlockingClassifier {
async fn classify(
&self,
_pane_text: &str,
) -> Result<(ActivityVerdict, u32, u32), ActivityError> {
self.entered.notify_one();
self.release.notified().await;
Ok((
ActivityVerdict {
state: ActivityState::Working,
summary: "blocked".into(),
confidence: 1.0,
},
1,
1,
))
}
}
let entered = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
let monitor = Arc::new(ActivityMonitor::new(
BlockingClassifier {
entered: entered.clone(),
release: release.clone(),
},
"test-model",
));
let entered_sub = entered.notified();
tokio::pin!(entered_sub);
entered_sub.as_mut().enable();
let m = monitor.clone();
let handle = tokio::spawn(async move { m.check("s1", "pane").await });
entered_sub.await;
assert!(monitor.is_summarizing("s1"));
release.notify_one();
handle.await.unwrap().unwrap();
assert!(!monitor.is_summarizing("s1"));
assert_eq!(monitor.cached_summary("s1").as_deref(), Some("blocked"));
}
#[tokio::test]
async fn monitor_different_sessions_independent_caches() {
let classifier = MockClassifier::new(ActivityState::Working);
let monitor = ActivityMonitor::new(classifier, "test-model");
monitor.check("s1", "content A").await.unwrap();
monitor.check("s2", "content B").await.unwrap();
assert_eq!(monitor.llm.calls().await, 2);
}
#[test]
fn sha256_hex_is_deterministic() {
let h1 = sha256_hex(b"hello");
let h2 = sha256_hex(b"hello");
assert_eq!(h1, h2);
assert_ne!(h1, sha256_hex(b"world"));
}
#[test]
fn last_n_lines_short() {
let text = "a\nb\nc";
assert_eq!(last_n_lines(text, 10), "a\nb\nc");
}
#[test]
fn last_n_lines_long() {
let text = (0..100)
.map(|i| i.to_string())
.collect::<Vec<_>>()
.join("\n");
let tail = last_n_lines(&text, 60);
let lines: Vec<&str> = tail.lines().collect();
assert_eq!(lines.len(), 60);
assert_eq!(lines[0], "40");
}
struct FailingClassifier {
missing_key: bool,
}
impl LlmClassifier for FailingClassifier {
async fn classify(
&self,
_pane_text: &str,
) -> Result<(ActivityVerdict, u32, u32), ActivityError> {
if self.missing_key {
Err(ActivityError::MissingApiKey)
} else {
Err(ActivityError::Llm("provider exploded".into()))
}
}
}
#[tokio::test]
async fn missing_api_key_degrades_to_unknown_verdict() {
let monitor = ActivityMonitor::new(FailingClassifier { missing_key: true }, "test-model");
let result = monitor
.check("s1", "pane")
.await
.expect("degrades, not errors");
assert_eq!(result.verdict.state, ActivityState::Unknown);
assert_eq!(result.verdict.summary, "OPENROUTER_API_KEY not configured");
assert_eq!(
(result.cost.input_tokens, result.cost.output_tokens),
(0, 0)
);
assert_eq!(monitor.cached_summary("s1"), None);
assert!(!monitor.is_summarizing("s1"));
}
#[tokio::test]
async fn classifier_error_propagates() {
let monitor = ActivityMonitor::new(FailingClassifier { missing_key: false }, "test-model");
let err = monitor.check("s1", "pane").await.expect_err("propagates");
assert!(matches!(err, ActivityError::Llm(_)), "got {err:?}");
assert!(!monitor.is_summarizing("s1"));
}
}