feat(agent): expose data quality in portal

adk-rust/crates/awatch-agent-rs/src/transport.rs

@@ -171,6 +171,15 @@ impl AwWorktimePublisher {
         let mut sent = 0;
         let sample_seconds = 60_i64;
         for session in sessions {
+            let ignore_for_kpi = ignored_for_kpi(record, &session);
+            let active_for_kpi = session.active && !ignore_for_kpi;
+            let state = if ignore_for_kpi {
+                "IgnoredForKpi"
+            } else if session.active {
+                "Active"
+            } else {
+                "Disconnected"
+            };
             let payload = serde_json::json!({
                 "timestamp": record.timestamp,
                 "duration": sample_seconds,
@@ -180,8 +189,10 @@ impl AwWorktimePublisher {
                     "sessionId": session_id_number(&session),
                     "sessionName": session.session_type,
                     "sessionSource": session.session_source,
-                    "state": if session.active { "Active" } else { "Disconnected" },
+                    "state": state,
-                    "active": session.active,
+                    "active": active_for_kpi,
+                    "ignoredForKpi": ignore_for_kpi,
+                    "qualityNote": if ignore_for_kpi { Some("local_fallback is diagnostics-only and is not accepted as activity proof") } else { None },
                     "sampleSeconds": sample_seconds,
                     "pollSeconds": sample_seconds,
                     "hostname": record.hostname,
@@ -227,6 +238,11 @@ impl AwWorktimePublisher {
     }
 }
 
+fn ignored_for_kpi(record: &TelemetryRecord, session: &SessionInfo) -> bool {
+    record.diagnostics.collector_source == "local_fallback"
+        || session.session_source.as_deref() == Some("local_fallback")
+}
+
 fn ensure_aw_bucket(
     client: &Client,
     aw_api_base: &str,
@@ -429,6 +445,30 @@ mod tests {
         assert_eq!(session_id_number(&session), 12);
     }
 
+    #[test]
+    fn local_fallback_sessions_are_diagnostics_only_for_kpi() {
+        let mut wts_record = record();
+        wts_record.diagnostics = diagnostics_for_sessions(&[], &[], "wts_api", None);
+        let session = SessionInfo {
+            session_id: "0".to_string(),
+            username: "user".to_string(),
+            session_type: "local".to_string(),
+            session_source: Some("local_fallback".to_string()),
+            remote_addr: None,
+            started_at: None,
+            active: true,
+        };
+        assert!(ignored_for_kpi(&wts_record, &session));
+
+        let mut fallback_record = record();
+        fallback_record.diagnostics = diagnostics_for_sessions(&[], &[], "local_fallback", None);
+        let session = SessionInfo {
+            session_source: Some("wts_api".to_string()),
+            ..session
+        };
+        assert!(ignored_for_kpi(&fallback_record, &session));
+    }
+
     #[test]
     fn worktime_publisher_is_disabled_by_default() {
         assert!(AwWorktimePublisher::new(&AgentConfig::default()).is_none());

adk-rust/crates/detmir-portal/src/main.rs

@@ -165,6 +165,30 @@ struct SourceStatus {
     payload: Option<Value>,
 }
 
+#[derive(Clone, Debug, Serialize)]
+struct AgentQuality {
+    collector_source: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    collector_error: Option<String>,
+    sessions_collected_total: usize,
+    active_sessions_total: usize,
+    rdp_sessions_total: usize,
+    quality_status: String,
+}
+
+impl Default for AgentQuality {
+    fn default() -> Self {
+        Self {
+            collector_source: "unknown".to_string(),
+            collector_error: None,
+            sessions_collected_total: 0,
+            active_sessions_total: 0,
+            rdp_sessions_total: 0,
+            quality_status: "unknown".to_string(),
+        }
+    }
+}
+
 #[derive(Debug, Serialize)]
 struct HealthResponse {
     ok: bool,
@@ -402,6 +426,7 @@ struct Snapshot {
     worktime: SourceStatus,
     worktime_management: SourceStatus,
     one_c: SourceStatus,
+    agent_quality: AgentQuality,
 }
 
 #[derive(Debug)]
@@ -935,9 +960,92 @@ fn build_snapshot(args: &Cli) -> Snapshot {
             &format!("{}/api/health", args.one_c_url.trim_end_matches('/')),
             timeout,
         ),
+        agent_quality: load_agent_quality(&args.telemetry_store_path),
     }
 }
 
+fn load_agent_quality(path: &Path) -> AgentQuality {
+    let Some(payload) = latest_telemetry_record(path) else {
+        return AgentQuality::default();
+    };
+    agent_quality_from_record(&payload)
+}
+
+fn latest_telemetry_record(path: &Path) -> Option<Value> {
+    let text = fs::read_to_string(path).ok()?;
+    text.lines().rev().find_map(|line| {
+        let envelope = serde_json::from_str::<Value>(line).ok()?;
+        envelope.get("record").cloned().or(Some(envelope))
+    })
+}
+
+fn agent_quality_from_record(record: &Value) -> AgentQuality {
+    let Some(diagnostics) = record.get("diagnostics").and_then(Value::as_object) else {
+        return AgentQuality::default();
+    };
+    let collector_source = diagnostics
+        .get("collector_source")
+        .and_then(Value::as_str)
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or("unknown")
+        .to_string();
+    let collector_error = diagnostics
+        .get("collector_error")
+        .and_then(Value::as_str)
+        .filter(|value| !value.trim().is_empty())
+        .map(str::to_string);
+    AgentQuality {
+        sessions_collected_total: diagnostics
+            .get("sessions_collected_total")
+            .and_then(Value::as_u64)
+            .unwrap_or(0) as usize,
+        active_sessions_total: diagnostics
+            .get("active_sessions_total")
+            .and_then(Value::as_u64)
+            .unwrap_or(0) as usize,
+        rdp_sessions_total: diagnostics
+            .get("rdp_sessions_total")
+            .and_then(Value::as_u64)
+            .unwrap_or(0) as usize,
+        quality_status: agent_quality_status(&collector_source, collector_error.as_deref())
+            .to_string(),
+        collector_source,
+        collector_error,
+    }
+}
+
+fn agent_quality_status(collector_source: &str, collector_error: Option<&str>) -> &'static str {
+    if let Some(error) = collector_error.filter(|value| !value.trim().is_empty()) {
+        return if critical_collector_error(error) {
+            "error"
+        } else {
+            "degraded"
+        };
+    }
+    match collector_source {
+        "wts_api" => "ok",
+        "quser_utf16" | "quser_lossy" | "env_sessionname_fallback" => "fallback",
+        "local_fallback" => "degraded",
+        _ => "unknown",
+    }
+}
+
+fn critical_collector_error(error: &str) -> bool {
+    let lower = error.to_lowercase();
+    [
+        "access denied",
+        "permission",
+        "unauthorized",
+        "invalid",
+        "panic",
+        "cannot parse",
+        "failed to parse",
+        "missing required",
+    ]
+    .iter()
+    .any(|needle| lower.contains(needle))
+}
+
 fn command_json_source(name: &str, command: &str, timeout: Duration) -> SourceStatus {
     match run_shell(command, timeout) {
         Ok((stdout, stderr, success)) => {
@@ -1260,6 +1368,7 @@ fn build_reports(
     };
     let grafana = grafana_block(snapshot);
     let collection = collection_block(snapshot.detmir_check.payload.as_ref());
+    let agent_quality = snapshot.agent_quality.clone();
     let worktime = worktime_block(snapshot);
     let one_c = one_c_block(snapshot);
     let dlp_block_value = dlp_block(snapshot);
@@ -1288,6 +1397,10 @@ fn build_reports(
     };
     let executive_points = vec![
         format!("Сбор данных: {}. {}", collection.status, collection.text),
+        format!(
+            "Качество данных агента: {} через {}",
+            agent_quality.quality_status, agent_quality.collector_source
+        ),
         format!(
             "Работа сегодня: сотрудников={}, активное время={}",
             metrics.users_count,
@@ -1332,6 +1445,7 @@ fn build_reports(
         "executive_points": executive_points,
         "kpis": [
             report_kpi("UEBA риск", format!("{}/100", ueba_risk.get("score").and_then(Value::as_u64).unwrap_or(0)), ueba_risk.get("status").and_then(Value::as_str).unwrap_or("UNKNOWN").to_string(), ueba_risk.get("summary").and_then(Value::as_str).unwrap_or("risk score")),
+            report_kpi("Качество данных агента", agent_quality.quality_status.clone(), agent_quality.quality_status.clone(), &format!("источник: {}", agent_quality.collector_source)),
             report_kpi("Индекс активности", workforce_index_text(metrics.workforce_index), workforce_index_status(metrics.workforce_index), "proxy: активное время / плановое рабочее время"),
             weighted_activity_kpi_from_policy(&workforce_policy_explain),
             report_kpi("Сотрудники", metrics.users_count.to_string(), worktime.status.clone(), "строки worktime за сегодня"),
@@ -1348,6 +1462,7 @@ fn build_reports(
                 "items": [
                     report_item("DetMir status", snapshot.detmir_status.status.clone(), snapshot.detmir_status.summary.clone()),
                     report_item("Сбор данных", collection.status.clone(), collection.text.clone()),
+                    report_item("Качество данных агента", agent_quality.quality_status.clone(), format!("source={}, sessions={}, active={}, rdp={}", agent_quality.collector_source, agent_quality.sessions_collected_total, agent_quality.active_sessions_total, agent_quality.rdp_sessions_total)),
                     report_item("Grafana", grafana.status.clone(), grafana.text.clone()),
                     report_item("1C analytics", one_c.status.clone(), one_c.text.clone())
                 ]
@@ -1396,6 +1511,7 @@ fn build_reports(
         ],
         "ueba_risk": ueba_risk,
         "ueba_baseline": ueba_baseline,
+        "agent_quality": agent_quality,
         "workforce_policy": workforce_policy_explain,
         "workforce": {
             "department_comparison": department_items,
@@ -2791,6 +2907,19 @@ fn render_report_markdown(
         "- Индекс активности: {}\n",
         workforce_index_text(metrics.workforce_index)
     ));
+    text.push_str(&format!(
+        "- Качество данных агента: {} через {}\n",
+        snapshot.agent_quality.quality_status, snapshot.agent_quality.collector_source
+    ));
+    text.push_str(&format!(
+        "- Сессии агента: всего={}, активные={}, RDP={}\n",
+        snapshot.agent_quality.sessions_collected_total,
+        snapshot.agent_quality.active_sessions_total,
+        snapshot.agent_quality.rdp_sessions_total
+    ));
+    if let Some(error) = &snapshot.agent_quality.collector_error {
+        text.push_str(&format!("- Ошибка коллектора агента: {error}\n"));
+    }
     text.push_str(&format!(
         "- Сотрудники за сегодня: {}\n",
         metrics.users_count
@@ -4676,6 +4805,57 @@ mod tests {
         assert_eq!(block.status, "OK");
     }
 
+    #[test]
+    fn agent_quality_status_prioritizes_collector_risk() {
+        assert_eq!(agent_quality_status("wts_api", None), "ok");
+        assert_eq!(agent_quality_status("quser_utf16", None), "fallback");
+        assert_eq!(agent_quality_status("quser_lossy", None), "fallback");
+        assert_eq!(
+            agent_quality_status("env_sessionname_fallback", None),
+            "fallback"
+        );
+        assert_eq!(agent_quality_status("local_fallback", None), "degraded");
+        assert_eq!(agent_quality_status("unknown", None), "unknown");
+        assert_eq!(
+            agent_quality_status("wts_api", Some("temporary query failure")),
+            "degraded"
+        );
+        assert_eq!(
+            agent_quality_status("wts_api", Some("access denied by WTS API")),
+            "error"
+        );
+    }
+
+    #[test]
+    fn agent_quality_defaults_to_unknown_for_old_payloads() {
+        let quality = agent_quality_from_record(&json!({
+            "agent_id": "agent-legacy",
+            "hostname": "HOST-EXAMPLE"
+        }));
+        assert_eq!(quality.quality_status, "unknown");
+        assert_eq!(quality.collector_source, "unknown");
+        assert_eq!(quality.sessions_collected_total, 0);
+    }
+
+    #[test]
+    fn agent_quality_loads_latest_jsonl_diagnostics() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("telemetry.jsonl");
+        fs::write(
+            &path,
+            r#"{"record":{"agent_id":"old","diagnostics":{"collector_source":"local_fallback","sessions_collected_total":1,"active_sessions_total":1,"rdp_sessions_total":0}}}
+{"record":{"agent_id":"new","diagnostics":{"collector_source":"wts_api","sessions_collected_total":3,"active_sessions_total":2,"rdp_sessions_total":2}}}
+"#,
+        )
+        .unwrap();
+        let quality = load_agent_quality(&path);
+        assert_eq!(quality.quality_status, "ok");
+        assert_eq!(quality.collector_source, "wts_api");
+        assert_eq!(quality.sessions_collected_total, 3);
+        assert_eq!(quality.active_sessions_total, 2);
+        assert_eq!(quality.rdp_sessions_total, 2);
+    }
+
     #[test]
     fn incident_id_is_stable() {
         assert_eq!(
@@ -4996,6 +5176,7 @@ mod tests {
                 error: None,
                 payload: Some(json!({"status": "ok", "companies_total": 47})),
             },
+            agent_quality: AgentQuality::default(),
         };
         let evidence = DlpEvidenceResponse {
             ok: true,
@@ -5080,6 +5261,13 @@ mod tests {
         assert_eq!(report["workforce_policy"]["configured"], false);
         assert_eq!(report["workforce"]["trend_status"], "daily_only");
         assert_eq!(report["workforce"]["insights"].as_array().unwrap().len(), 1);
+        assert_eq!(report["agent_quality"]["quality_status"], "unknown");
+        assert!(
+            report["markdown"]
+                .as_str()
+                .unwrap()
+                .contains("Качество данных агента")
+        );
         assert_eq!(
             report["workforce"]["department_comparison"]
                 .as_array()
@@ -5154,6 +5342,7 @@ confidence:
                 error: None,
                 payload: None,
             },
+            agent_quality: AgentQuality::default(),
         };
         let metrics = ReportMetrics {
             users_count: 1,
@@ -5256,6 +5445,7 @@ confidence:
                     error: None,
                     payload: None,
                 },
+                agent_quality: AgentQuality::default(),
             }
         }
 
@@ -5343,6 +5533,7 @@ confidence:
                 error: None,
                 payload: None,
             },
+            agent_quality: AgentQuality::default(),
         };
         let policy = WorkforcePolicy {
             default_role: "accountant".to_string(),
@@ -5441,6 +5632,7 @@ confidence:
                 error: None,
                 payload: None,
             },
+            agent_quality: AgentQuality::default(),
         };
         let explain = build_workforce_policy_explain(&snapshot, &policy_path, false);
         assert_eq!(explain["configured"], true);

adk-rust/crates/detmir-portal/src/static/app.css

@@ -9,6 +9,8 @@
   --ok-bg: #dcfce7;
   --warn: #f97316;
   --warn-bg: #ffedd5;
+  --degraded: #ea580c;
+  --degraded-bg: #fed7aa;
   --fail: #dc2626;
   --fail-bg: #fee2e2;
   --unknown: #64748b;
@@ -262,6 +264,7 @@ h1 {
 
 .status-ok { background: var(--ok-bg); color: var(--ok); }
 .status-warn { background: var(--warn-bg); color: var(--warn); }
+.status-degraded { background: var(--degraded-bg); color: var(--degraded); }
 .status-fail { background: var(--fail-bg); color: var(--fail); }
 .status-unknown { background: var(--unknown-bg); color: var(--unknown); }
 .text-ok { color: var(--ok); }
@@ -504,6 +507,49 @@ h1 {
   line-height: 1.35;
 }
 
+.agent-quality-card {
+  margin: 12px 0 16px;
+}
+
+.quality-grid {
+  display: grid;
+  grid-template-columns: repeat(auto-fit, minmax(150px, 1fr));
+  gap: 10px;
+  margin-top: 10px;
+}
+
+.quality-grid > div {
+  display: grid;
+  gap: 4px;
+  min-width: 0;
+  padding: 10px;
+  border: 1px solid var(--line);
+  border-radius: 8px;
+  background: var(--soft);
+}
+
+.quality-grid strong {
+  overflow-wrap: anywhere;
+}
+
+.quality-warning {
+  margin-top: 10px;
+  padding: 10px 12px;
+  border-left: 4px solid var(--warn);
+  border-radius: 8px;
+  background: var(--warn-bg);
+  color: #9a3412;
+  font-weight: 700;
+  line-height: 1.35;
+}
+
+.quality-error {
+  margin: 10px 0 0;
+  color: var(--fail);
+  font-weight: 700;
+  overflow-wrap: anywhere;
+}
+
 .heatmap-table td {
   min-width: 120px;
 }

adk-rust/crates/detmir-portal/src/static/app.js

@@ -24,8 +24,9 @@ async function postJson(path, payload) {
 function statusClass(status) {
   const s = String(status || "UNKNOWN").toLowerCase();
   if (s === "ok" || s === "true") return "status-ok";
-  if (s === "warn" || s === "warning") return "status-warn";
+  if (s === "warn" || s === "warning" || s === "fallback") return "status-warn";
-  if (s === "fail" || s === "false") return "status-fail";
+  if (s === "degraded") return "status-degraded";
+  if (s === "fail" || s === "false" || s === "error") return "status-fail";
   return "status-unknown";
 }
 
@@ -805,6 +806,7 @@ function renderOperator(data, report) {
     </div>
     ${renderPeriodBanner(report)}
     ${renderExecutiveMetrics(report, incidents)}
+    ${renderAgentQuality(report?.agent_quality)}
     ${renderOverviewAnalytics(report)}
     <section class="dashboard-band">
       <div class="band-head"><h3>Рабочая активность сотрудников</h3><span class="muted">загрузка, простои, перегруз и дисциплина процессов</span></div>
@@ -1334,6 +1336,32 @@ function renderUebaRisk(risk) {
   `;
 }
 
+function renderAgentQuality(quality) {
+  const q = quality || {};
+  const status = q.quality_status || "unknown";
+  const source = q.collector_source || "unknown";
+  const warn = ["fallback", "degraded", "error"].includes(String(status).toLowerCase());
+  return `
+    <section class="card agent-quality-card">
+      <div class="section-head">
+        <div>
+          <h3>Качество данных агента</h3>
+          <p class="muted">Доверие к источнику, который подтверждает активность и RDP-сессии.</p>
+        </div>
+        <span class="badge ${statusClass(status)}">${ui(status)}</span>
+      </div>
+      ${warn ? `<div class="quality-warning">Внимание. Данные активности собраны не основным способом. Точность определения активности и RDP-сессий может быть снижена.</div>` : ""}
+      <div class="quality-grid">
+        <div><span class="muted">Источник</span><strong>${ui(source)}</strong></div>
+        <div><span class="muted">Сессий собрано</span><strong>${escapeHtml(q.sessions_collected_total ?? 0)}</strong></div>
+        <div><span class="muted">Активных сессий</span><strong>${escapeHtml(q.active_sessions_total ?? 0)}</strong></div>
+        <div><span class="muted">RDP-сессий</span><strong>${escapeHtml(q.rdp_sessions_total ?? 0)}</strong></div>
+      </div>
+      ${q.collector_error ? `<p class="quality-error">Ошибка коллектора: ${ui(q.collector_error)}</p>` : ""}
+    </section>
+  `;
+}
+
 function renderReports(data) {
   data = periodReport(data);
   return `
@@ -1368,6 +1396,7 @@ function renderReports(data) {
     </div>
     <h3 class="section-title">Ключевые показатели</h3>
     ${renderKpiCards(data.kpis)}
+    ${renderAgentQuality(data.agent_quality)}
     ${renderUebaRisk(data.ueba_risk)}
     ${renderWorkforceIndexExplanation(data.workforce_policy)}
     <h3 class="section-title">Срезы отчета</h3>

docs/PORTAL_RU.md

@@ -28,10 +28,37 @@
 - WARN/FAIL подразделения;
 - критические риски;
 - недельный тренд;
+- качество данных агента;
 - топ-5 лучших и проблемных подразделений;
 - Heat Map;
 - блок `Требует внимания`.
 
+## Качество данных агента
+
+Портал показывает карточку `Качество данных агента` в `Обзор` и `Отчеты`.
+
+Карточка содержит:
+
+- `quality_status`;
+- `collector_source`;
+- `sessions_collected_total`;
+- `active_sessions_total`;
+- `rdp_sessions_total`;
+- `collector_error`, если он есть.
+
+Цвета статусов:
+
+- `ok` - зеленый;
+- `fallback` - желтый;
+- `degraded` - оранжевый;
+- `error` - красный;
+- `unknown` - серый.
+
+Для `fallback`, `degraded` и `error` портал показывает предупреждение о том,
+что активность собрана не основным способом и точность RDP/worktime KPI может
+быть снижена. Старые агенты без diagnostics не ломают API и отображаются как
+`unknown`.
+
 ## Risk -> Investigation
 
 Кнопка `Открыть расследование` переводит пользователя в read-only карточку.

docs/WINDOWS_RUST_AGENT_WORKTIME_RU.md

@@ -73,6 +73,26 @@ aw_worktime_enabled = true
 - `collector_source`;
 - `collector_error`.
 
+## Качество данных агента
+
+Портал и отчеты поднимают diagnostics в блок `agent_quality`.
+
+Статусы:
+
+- `ok` - основной источник `wts_api`, ошибки коллектора нет;
+- `fallback` - данные получены через `quser_utf16`, `quser_lossy` или
+  `env_sessionname_fallback`;
+- `degraded` - используется `local_fallback` или есть некритичная ошибка
+  коллектора;
+- `error` - есть критичная ошибка коллектора, например отказ доступа,
+  некорректный обязательный payload или ошибка парсинга;
+- `unknown` - старый агент или payload без diagnostics.
+
+`local_fallback` считается диагностическим сигналом, а не доказательством
+активности. События worktime, опубликованные из `local_fallback`, получают
+`active=false`, `ignoredForKpi=true` и не должны увеличивать KPI сотрудника или
+подтверждать RDP-активность.
+
 ## PowerShell Legacy Fallback
 
 В `deployment-config.json` используется блок:
@@ -143,7 +163,9 @@ curl "http://<AW_SERVER_HOST>:5600/api/0/buckets/aw-worktime-sessions_<HOST>/eve
 ```json
 {
   "source": "awatch-agent-rs",
-  "sessionSource": "wts_api"
+  "sessionSource": "wts_api",
+  "collectorSource": "wts_api",
+  "ignoredForKpi": false
 }
 ```
 

scripts/detmir-portal-tabs-smoke.mjs

@@ -120,6 +120,7 @@ async function main() {
         const requiredExecutive = [
           "Сотрудников в работе",
           "Средний индекс активности",
+          "Качество данных агента",
           "ТОП-5 лучших подразделений",
           "ТОП-5 проблемных подразделений",
           "Требует внимания",