Fixed agent feedback

Author: kirit93

crates/openshell-core/src/telemetry.rs

@@ -6,11 +6,13 @@
 use chrono::{SecondsFormat, Utc};
 use reqwest::blocking::Client;
 use serde_json::{Value, json};
+use std::collections::BTreeMap;
 use std::sync::{OnceLock, mpsc};
 use std::thread;
 use std::time::Duration;
 
 const TELEMETRY_EVENT_QUEUE_CAPACITY: usize = 1024;
+const MAX_TELEMETRY_INTEGER: u64 = 9_223_372_036_854_775_807;
 const CLIENT_ID: &str = "415437562476676";
 const DEFAULT_ENDPOINT: &str = "https://events.telemetry.data-uat.nvidia.com/v1.1/events/json";
 const EVENT_SCHEMA_VERSION: &str = "2.0";
@@ -24,10 +26,11 @@ static TELEMETRY_SENDER: OnceLock<Option<mpsc::SyncSender<TelemetryEvent>>> = On
 struct TelemetryEvent {
     endpoint: String,
     name: &'static str,
+    event_ts: String,
     event: Value,
 }
 
-fn telemetry_enabled() -> bool {
+pub fn enabled() -> bool {
     telemetry_enabled_from(std::env::var("OPENSHELL_TELEMETRY_ENABLED").ok().as_deref())
 }
 
@@ -61,7 +64,7 @@ fn timestamp() -> String {
     Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
 }
 
-fn build_payload(name: &str, event: Value, ts: &str) -> Value {
+fn build_payload(name: &str, event: Value, event_ts: &str, sent_ts: &str) -> Value {
     json!({
         "browserType": "undefined",
         "clientId": CLIENT_ID,
@@ -90,12 +93,12 @@ fn build_payload(name: &str, event: Value, ts: &str) -> Value {
         "integrationId": "undefined",
         "productName": "undefined",
         "productVersion": "undefined",
-        "sentTs": ts,
+        "sentTs": sent_ts,
         "sessionId": "undefined",
         "userId": "undefined",
         "events": [
             {
-                "ts": ts,
+                "ts": event_ts,
                 "parameters": event,
                 "name": name,
             }
@@ -118,7 +121,7 @@ fn telemetry_sender() -> Option<&'static mpsc::SyncSender<TelemetryEvent>> {
 
 fn telemetry_worker(rx: mpsc::Receiver<TelemetryEvent>) {
     for event in rx {
-        let payload = build_payload(event.name, event.event, &timestamp());
+        let payload = build_payload(event.name, event.event, &event.event_ts, &timestamp());
         let _ = publish_payload(&event.endpoint, payload);
     }
 }
@@ -141,7 +144,7 @@ fn try_enqueue_event(sender: &mpsc::SyncSender<TelemetryEvent>, event: Telemetry
 }
 
 fn emit_event(name: &'static str, event: Value) {
-    if !telemetry_enabled() {
+    if !enabled() {
         return;
     }
     let Some(endpoint) = telemetry_endpoint() else {
@@ -156,12 +159,22 @@ fn emit_event(name: &'static str, event: Value) {
         TelemetryEvent {
             endpoint,
             name,
+            event_ts: timestamp(),
             event,
         },
     );
 }
 
 pub fn emit_lifecycle(resource: &str, operation: &str, outcome: &str) {
+    let Some(resource) = lifecycle_resource(resource) else {
+        return;
+    };
+    let Some(operation) = lifecycle_operation(operation) else {
+        return;
+    };
+    let Some(outcome) = telemetry_outcome(outcome) else {
+        return;
+    };
     emit_event(
         "openshell_lifecycle_event",
         json!({
@@ -174,6 +187,13 @@ pub fn emit_lifecycle(resource: &str, operation: &str, outcome: &str) {
 }
 
 pub fn emit_provider_lifecycle(operation: &str, outcome: &str, provider_profile: &str) {
+    let Some(operation) = lifecycle_operation(operation) else {
+        return;
+    };
+    let Some(outcome) = telemetry_outcome(outcome) else {
+        return;
+    };
+    let provider_profile = provider_profile_bucket(provider_profile);
     emit_event(
         "openshell_provider_lifecycle_event",
         json!({
@@ -192,6 +212,13 @@ pub fn emit_sandbox_create(
     has_custom_policy: bool,
     template_source: &str,
 ) {
+    let Some(outcome) = telemetry_outcome(outcome) else {
+        return;
+    };
+    if !valid_count(provider_count) {
+        return;
+    }
+    let template_source = sandbox_template_source_bucket(template_source);
     emit_event(
         "openshell_sandbox_create_event",
         json!({
@@ -206,6 +233,15 @@ pub fn emit_sandbox_create(
 }
 
 pub fn emit_policy_decision(operation: &str, outcome: &str, rule_count: u64) {
+    let Some(operation) = policy_decision_operation(operation) else {
+        return;
+    };
+    let Some(outcome) = telemetry_outcome(outcome) else {
+        return;
+    };
+    if !valid_count(rule_count) {
+        return;
+    }
     emit_event(
         "openshell_policy_decision_event",
         json!({
@@ -226,21 +262,20 @@ pub fn emit_sandbox_activity_summary<I, S>(
     I: IntoIterator<Item = (S, u64)>,
     S: Into<String>,
 {
-    let mut rows: Vec<Value> = denials_by_group
+    if !valid_count(network_activity_count)
+        || !valid_count(denied_action_count)
+        || !denial_rate_pct.is_finite()
+        || !(0.0..=100.0).contains(&denial_rate_pct)
+    {
+        return;
+    }
+    let Some(denials_by_group) = sanitize_denials_by_group(denials_by_group) else {
+        return;
+    };
+    let rows: Vec<Value> = denials_by_group
         .into_iter()
-        .map(|(group, count)| {
-            json!({
-                "denyGroup": group.into(),
-                "deniedCount": count,
-            })
-        })
+        .map(|(group, count)| json!({ "denyGroup": group, "deniedCount": count }))
         .collect();
-    rows.sort_by(|left, right| {
-        left["denyGroup"]
-            .as_str()
-            .unwrap_or_default()
-            .cmp(right["denyGroup"].as_str().unwrap_or_default())
-    });
     emit_event(
         "openshell_sandbox_activity_summary_event",
         json!({
@@ -253,6 +288,107 @@ pub fn emit_sandbox_activity_summary<I, S>(
     );
 }
 
+fn valid_count(value: u64) -> bool {
+    value <= MAX_TELEMETRY_INTEGER
+}
+
+fn telemetry_outcome(raw: &str) -> Option<&'static str> {
+    match raw {
+        "success" => Some("success"),
+        "failure" => Some("failure"),
+        _ => None,
+    }
+}
+
+fn lifecycle_resource(raw: &str) -> Option<&'static str> {
+    match raw {
+        "sandbox" => Some("sandbox"),
+        "sandbox_policy" => Some("sandbox_policy"),
+        _ => None,
+    }
+}
+
+fn lifecycle_operation(raw: &str) -> Option<&'static str> {
+    match raw {
+        "create" => Some("create"),
+        "delete" => Some("delete"),
+        "update" => Some("update"),
+        _ => None,
+    }
+}
+
+fn policy_decision_operation(raw: &str) -> Option<&'static str> {
+    match raw {
+        "approve" => Some("approve"),
+        "reject" => Some("reject"),
+        "undo" => Some("undo"),
+        "approve_all" => Some("approve_all"),
+        _ => None,
+    }
+}
+
+fn sandbox_template_source_bucket(raw: &str) -> &'static str {
+    match raw {
+        "default" => "default",
+        "image" => "image",
+        _ => "undefined",
+    }
+}
+
+fn provider_profile_bucket(raw: &str) -> &'static str {
+    match raw.trim().to_ascii_lowercase().as_str() {
+        "anthropic" => "anthropic",
+        "claude" => "claude",
+        "codex" => "codex",
+        "copilot" => "copilot",
+        "github" => "github",
+        "gitlab" => "gitlab",
+        "nvidia" => "nvidia",
+        "openai" => "openai",
+        "opencode" => "opencode",
+        "outlook" => "outlook",
+        _ => "custom",
+    }
+}
+
+fn deny_group_bucket(raw: &str) -> &'static str {
+    match raw {
+        "connect_policy" | "connect" | "l4_deny" => "connect_policy",
+        "forward_policy" | "forward" => "forward_policy",
+        "l7_policy" | "l7" | "l7_deny" | "forward-l7-deny" => "l7_policy",
+        "l7_parse_rejection" | "parse_rejection" => "l7_parse_rejection",
+        "ssrf" => "ssrf",
+        "bypass" => "bypass",
+        "policy_stale" => "policy_stale",
+        _ => "unknown",
+    }
+}
+
+fn sanitize_denials_by_group<I, S>(denials_by_group: I) -> Option<BTreeMap<&'static str, u64>>
+where
+    I: IntoIterator<Item = (S, u64)>,
+    S: Into<String>,
+{
+    let mut sanitized = BTreeMap::<&'static str, u64>::new();
+    for (group, count) in denials_by_group {
+        if !valid_count(count) {
+            return None;
+        }
+        let group = group.into();
+        let bucket = deny_group_bucket(&group);
+        let next_count = sanitized
+            .get(bucket)
+            .copied()
+            .unwrap_or(0)
+            .checked_add(count)?;
+        if !valid_count(next_count) {
+            return None;
+        }
+        sanitized.insert(bucket, next_count);
+    }
+    Some(sanitized)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -291,6 +427,7 @@ mod tests {
                 "templateSource": "default",
             }),
             "2026-05-18T00:00:00.000Z",
+            "2026-05-18T00:00:01.000Z",
         );
 
         assert_eq!(payload["clientId"], CLIENT_ID);
@@ -303,6 +440,7 @@ mod tests {
         );
         assert_eq!(payload["events"][0]["parameters"]["nvidiaSource"], SOURCE);
         assert_eq!(payload["events"][0]["ts"], "2026-05-18T00:00:00.000Z");
+        assert_eq!(payload["sentTs"], "2026-05-18T00:00:01.000Z");
     }
 
     #[test]
@@ -311,6 +449,7 @@ mod tests {
         let event = || TelemetryEvent {
             endpoint: "https://example.test/events".to_string(),
             name: "openshell_lifecycle_event",
+            event_ts: "2026-05-18T00:00:00.000Z".to_string(),
             event: json!({
                 "nvidiaSource": SOURCE,
                 "resource": "sandbox",
@@ -322,4 +461,36 @@ mod tests {
         assert!(try_enqueue_event(&tx, event()));
         assert!(!try_enqueue_event(&tx, event()));
     }
+
+    #[test]
+    fn telemetry_validation_maps_privacy_sensitive_strings_to_safe_buckets() {
+        assert_eq!(provider_profile_bucket("corp-llm-prod"), "custom");
+        assert_eq!(
+            sandbox_template_source_bucket("ghcr.io/acme/private:latest"),
+            "undefined"
+        );
+        assert_eq!(deny_group_bucket("host=private.example"), "unknown");
+    }
+
+    #[test]
+    fn telemetry_validation_rejects_schema_invalid_values() {
+        assert_eq!(lifecycle_resource("gateway"), None);
+        assert_eq!(lifecycle_operation("restart"), None);
+        assert_eq!(policy_decision_operation("merge_internal_rule"), None);
+        assert_eq!(telemetry_outcome("partial"), None);
+        assert!(!valid_count(MAX_TELEMETRY_INTEGER + 1));
+    }
+
+    #[test]
+    fn activity_groups_are_sanitized_and_aggregated() {
+        let rows = sanitize_denials_by_group([
+            ("connect".to_string(), 1),
+            ("connect_policy".to_string(), 2),
+            ("host=private.example".to_string(), 3),
+        ])
+        .expect("rows should sanitize");
+
+        assert_eq!(rows.get("connect_policy"), Some(&3));
+        assert_eq!(rows.get("unknown"), Some(&3));
+    }
 }

crates/openshell-sandbox/src/activity_aggregator.rs

@@ -10,6 +10,7 @@ use tracing::debug;
 
 pub const ACTIVITY_EVENT_QUEUE_CAPACITY: usize = 1024;
 const ACTIVITY_FLUSH_QUEUE_CAPACITY: usize = 1;
+pub const DEFAULT_ACTIVITY_FLUSH_INTERVAL_SECS: u64 = 10;
 
 #[derive(Debug, Clone)]
 pub struct ActivityEvent {
@@ -120,6 +121,13 @@ pub fn try_record_activity(tx: &ActivitySender, denied: bool, deny_group: &'stat
     tx.try_send(ActivityEvent { denied, deny_group }).is_ok()
 }
 
+pub fn activity_flush_interval_secs_from_env(value: Option<&str>) -> u64 {
+    value
+        .and_then(|value| value.parse::<u64>().ok())
+        .filter(|value| *value > 0)
+        .unwrap_or(DEFAULT_ACTIVITY_FLUSH_INTERVAL_SECS)
+}
+
 fn queue_flush_summary(
     tx: &mpsc::Sender<FlushableActivitySummary>,
     summary: FlushableActivitySummary,
@@ -195,4 +203,21 @@ mod tests {
         assert!(queue_flush_summary(&tx, summary.clone()));
         assert!(!queue_flush_summary(&tx, summary));
     }
+
+    #[test]
+    fn activity_flush_interval_uses_positive_values_only() {
+        assert_eq!(
+            activity_flush_interval_secs_from_env(None),
+            DEFAULT_ACTIVITY_FLUSH_INTERVAL_SECS
+        );
+        assert_eq!(
+            activity_flush_interval_secs_from_env(Some("not-a-number")),
+            DEFAULT_ACTIVITY_FLUSH_INTERVAL_SECS
+        );
+        assert_eq!(
+            activity_flush_interval_secs_from_env(Some("0")),
+            DEFAULT_ACTIVITY_FLUSH_INTERVAL_SECS
+        );
+        assert_eq!(activity_flush_interval_secs_from_env(Some("5")), 5);
+    }
 }