From d5cc64252400820652b700d7dfca3bf00d303278 Mon Sep 17 00:00:00 2001 From: sunlight798 <3281498087@qq.com> Date: Sun, 25 Oct 2026 16:12:10 +0800 Subject: [PATCH 1/2] feat(desktop): collect bounded session observations Record native HTTP requests and Memory/Scope actions with bounded retention, safe projection and cancellation/write certainty semantics. Expose trusted summary, reset and diagnostic preview IPC with generated types and bilingual contracts. Refs #2107 (05-A) --- desktop/README.md | 2 + desktop/README.zh.md | 2 + desktop/SECURITY.md | 3 + desktop/VALIDATION.md | 21 + desktop/src-tauri/build.rs | 3 + desktop/src-tauri/capabilities/main.json | 5 +- desktop/src-tauri/examples/export_ipc.rs | 12 + desktop/src-tauri/src/commands.rs | 28 + desktop/src-tauri/src/connections/session.rs | 371 ++++++++------ desktop/src-tauri/src/lib.rs | 11 +- desktop/src-tauri/src/session_observations.rs | 477 ++++++++++++++++++ desktop/src-tauri/src/transport/api.rs | 177 ++++--- desktop/src-tauri/tests/api.rs | 71 +++ desktop/src-tauri/tests/ipc.rs | 49 ++ desktop/src-tauri/tests/memory.rs | 177 +++++++ desktop/src-tauri/tests/observations.rs | 87 ++++ desktop/src-tauri/tests/session.rs | 16 + desktop/ui/src/generated/ipc.ts | 12 + desktop/ui/src/shared/ipc.ts | 5 + .../desktop-session-observations.md | 106 ++++ docs/en/development/meta.json | 3 +- docs/en/development/observability.md | 3 + .../desktop-session-observations.md | 92 ++++ docs/zh/development/meta.json | 3 +- docs/zh/development/observability.md | 3 + 25 files changed, 1508 insertions(+), 231 deletions(-) create mode 100644 desktop/src-tauri/src/session_observations.rs create mode 100644 desktop/src-tauri/tests/observations.rs create mode 100644 docs/en/development/desktop-session-observations.md create mode 100644 docs/zh/development/desktop-session-observations.md diff --git a/desktop/README.md b/desktop/README.md index ed89cb8337..5beee319c1 100644 --- a/desktop/README.md +++ b/desktop/README.md @@ -8,6 +8,8 @@ See [中文说明](README.zh.md), [security boundary](SECURITY.md), and [validat The sidebar provides Overview, Connections, Memories, and Observation, with Settings & diagnostics at the bottom. Overview is the initial page. Observation contains Runtime state, Current-Scope inventory & usage, and This session tabs. Their data views are explicitly unavailable until integrated; opening the page or switching tabs does not run connection checks or local diagnostics. English/Chinese and existing theme settings apply to Observation. +The native host collects a bounded, in-memory window of HTTP requests and Memory/Scope user operations independently of page navigation. Trusted typed IPC provides an allowlisted summary, reset and safe JSON preview. The window retains at most 512 terminal records; it preserves cancellation and unknown-write outcomes and contains no content or target identities. See the [session observation contract](../docs/en/development/desktop-session-observations.md) for measurement units, coverage, reset and failure isolation. The This session page and clipboard flow require separate UI integration. + ## Shared feedback and current context diff --git a/desktop/README.zh.md b/desktop/README.zh.md index 8f18fe2473..19782c92e9 100644 --- a/desktop/README.zh.md +++ b/desktop/README.zh.md @@ -10,6 +10,8 @@ 侧栏提供总览、连接、记忆、观察入口，“设置与诊断”位于侧栏底部。默认进入总览。“观察”包含运行状态、当前 Scope 库存与用量、本次会话三个标签页。数据视图接入前明确显示尚未可用；进入页面或切换标签不会执行连接检查或本机诊断。观察页面沿用现有中英文和主题设置。 +原生宿主独立于页面导航，采集有界的内存窗口，分别记录 HTTP 请求和 Memory/Scope 用户操作。可信的类型化 IPC 提供白名单摘要、重置和安全 JSON 预览。窗口最多保留 512 条终态记录，保留取消及未知写入的区别，不包含内容或目标身份。测量单位、覆盖范围、重置和故障隔离见[会话观察契约](../docs/zh/development/desktop-session-observations.md)。本次会话页面和剪贴板流程仍需独立的 UI 集成。 + ## 统一反馈与当前上下文 diff --git a/desktop/SECURITY.md b/desktop/SECURITY.md index 9ae502eb6c..07b4c90f0c 100644 --- a/desktop/SECURITY.md +++ b/desktop/SECURITY.md @@ -29,6 +29,9 @@ Typed adapters separate liveness, readiness, identity and capabilities. Native c | Complete request, including response body | 15 seconds | | Response body | 1 MiB, checked both by declared size and streamed bytes | | Concurrent reads per client | 4; excess fails as busy | +| Session observations | 512 terminal records shared across HTTP requests and native Memory/Scope actions; memory only | + +Session observation commands require the same trusted main-window boundary and work independently of connection state. Rust projects only fixed operation/unit/outcome enums, elapsed milliseconds, UTC timestamps, counts, HTTP status and existing safe error codes. It never serializes connection or write state for export; content, queries, credentials, endpoints, private paths and target identities never enter the collector. The preview serializes only that projection. Retention/reset and process-lifetime loss metadata expose incomplete coverage; nonblocking recording cannot change business results or suppress cancellation. HTTP response success and native write certainty are separate measurements. Installer/Ops and local diagnostic output are outside collection. The [session observation contract](../docs/en/development/desktop-session-observations.md) defines the complete field allowlist and coverage. Error responses are enum codes. Transport exception text, endpoint URLs, bodies and credentials are never returned or logged. No product logging, telemetry, crash uploader or persistent business cache is configured. The app does not read databases, start a Server or inspect process environments. Explicit local diagnostics verify a pinned absolute executable and digest, then run one of two native-defined diagnostic commands. Registration requires no version, and legacy version fields are ignored. Diagnostic responses must match the supported output schema. A subsequent fixed `--version` query collects bounded display metadata; failure, unsupported output or an unavailable version does not reject an otherwise valid diagnostic result. Each invocation has a combined 256 KiB output budget and a deadline. Windows helpers enter an owned kill-on-close Job before resuming. macOS runs a read-only snapshot of the digest-verified launcher in a private process group, which is terminated after completion, cancellation or timeout; raw output is projected to allowlisted status fields and discarded. Normal application exit closes diagnostic admission and waits for active worker process cleanup before Tauri exits. A local CLI pin does not authenticate its publisher or freeze its Python dependencies; the configured installation must be trusted. diff --git a/desktop/VALIDATION.md b/desktop/VALIDATION.md index be8a7b5fab..ce3e4adcb8 100644 --- a/desktop/VALIDATION.md +++ b/desktop/VALIDATION.md @@ -20,6 +20,27 @@ cargo test --locked --manifest-path desktop/src-tauri/Cargo.toml [Desktop CI](../.github/workflows/desktop.yml) also builds a Server wheel, runs real Server and CLI tests, checks Windows credentials and process containment, builds the installer, and exercises installation, the installed UI and uninstallation. Consult the PR checks for the exact commit being reviewed. Reports, logs and screenshots are generated under ignored `.artifacts/` and uploaded as Actions artifacts; do not copy them into the repository. +## Native session observation checks + +Windows x64 checks for the native session collector use Rust 1.95.0, Node 24.14.1 and pnpm 11.13.1. +`cargo fmt --check`, `cargo clippy --locked --all-targets -- -D warnings`, all 62 native tests and generated IPC +checking pass. The new regressions exercise real loopback HTTP sockets with synthetic responses, including +readiness 503, request versus user-operation counting, millisecond latency, cancellation, unknown writes, +invalid decoded receipts, reset during an in-flight write, bounded retention, safe projection and collector +contention/poisoning. Mock Tauri IPC checks exercise the actual capability configuration and reject untrusted +windows and origins; session commands also work without connection state. + +Desktop lint, TypeScript checking, all 98 UI tests and the frontend production build pass. The Desktop OpenAPI +contract checks pass without contract changes. Website lint and link validation pass, including both locale +versions of the [session observation contract](../docs/en/development/desktop-session-observations.md). +`uv lock --locked` and `uv run --locked pytest tests/test_desktop_contract.py -q` pass. + +Repository hooks pass except `ty check` on Windows, which reports POSIX-only attributes in unchanged Python files +under `src/powercontext/builtin/code/` and `tests/builtin/test_native_code.py`. The explicit Linux-target check, +`uv run --locked ty check --python-platform linux`, passes. The native tests do not qualify an actual Server, +installed application, clipboard interaction or macOS execution. The This session UI and installed copy flow +remain separate delivery work; native collection does not make the placeholder page operational. + ## Compatibility scope [compatibility.json](src-tauri/src/connections/compatibility.json) preserves the recorded Server sources, normalized OpenAPI contract digests and qualification-wheel digests for the historical profiles below. The current contract is qualified by `sqlite-atomic-1da99747-v1` below; ordinary builds require explicitly selecting it and checking the connection before business operations become available. The `sqlite-atomic-d9a78567-v1`, `sqlite-atomic-caf2a35a-v1`, `sqlite-atomic-683e2db0-v1` and `sqlite-atomic-cab054db-v1` records qualify their corresponding Server source deployments and do not match this bundled contract. Historical records are disabled in the selector, and saved unavailable selections remain visible until explicitly replaced. CI builds have their own artifact identities and do not change product qualification. diff --git a/desktop/src-tauri/build.rs b/desktop/src-tauri/build.rs index 7bc3a15a92..50e9dda36d 100644 --- a/desktop/src-tauri/build.rs +++ b/desktop/src-tauri/build.rs @@ -49,6 +49,9 @@ fn main() { "cancel_scope_reads", "default_scope", "select_scope", + "session_summary", + "reset_session_observations", + "preview_session_diagnostics", ]), )) .expect("desktop build configuration is invalid"); diff --git a/desktop/src-tauri/capabilities/main.json b/desktop/src-tauri/capabilities/main.json index 668f40f201..b523012ec5 100644 --- a/desktop/src-tauri/capabilities/main.json +++ b/desktop/src-tauri/capabilities/main.json @@ -21,6 +21,9 @@ "allow-list-scopes", "allow-cancel-scope-reads", "allow-default-scope", - "allow-select-scope" + "allow-select-scope", + "allow-session-summary", + "allow-reset-session-observations", + "allow-preview-session-diagnostics" ] } diff --git a/desktop/src-tauri/examples/export_ipc.rs b/desktop/src-tauri/examples/export_ipc.rs index 66623da867..ed32be2967 100644 --- a/desktop/src-tauri/examples/export_ipc.rs +++ b/desktop/src-tauri/examples/export_ipc.rs @@ -26,6 +26,10 @@ use powercontext_desktop::{ diagnostics::{DiagnosticItem, DiagnosticKind, DiagnosticReport, HostDiagnostic}, error::SafeError, ipc::FoundationInfo, + session_observations::{ + ObservationOutcome, ObservationUnit, ObservedOperation, OperationSummary, OutcomeCounts, + SessionCoverage, SessionObservation, SessionSummary, + }, transport::{ ApiFailure, AtomicMemoryDetail, MemorySaveResponse, MemorySearchHit, MemorySearchResponse, wire, @@ -72,6 +76,14 @@ fn main() { MemorySearchHit::decl(&config), MemorySearchResponse::decl(&config), AtomicMemoryDetail::decl(&config), + ObservationUnit::decl(&config), + ObservedOperation::decl(&config), + ObservationOutcome::decl(&config), + SessionObservation::decl(&config), + OutcomeCounts::decl(&config), + OperationSummary::decl(&config), + SessionCoverage::decl(&config), + SessionSummary::decl(&config), ] .into_iter() .chain(wire::declarations(&config)) diff --git a/desktop/src-tauri/src/commands.rs b/desktop/src-tauri/src/commands.rs index d6e3cfe1c1..82bd80ab3e 100644 --- a/desktop/src-tauri/src/commands.rs +++ b/desktop/src-tauri/src/commands.rs @@ -21,6 +21,7 @@ use crate::{ }, error::SafeError, ipc::authorize_window, + session_observations::{SessionObservations, SessionSummary}, transport::{ ApiFailure, wire::{ScopeDescriptor, ScopePage}, @@ -29,6 +30,33 @@ use crate::{ pub struct HostState { pub manager: Result, } + +#[tauri::command] +pub fn session_summary( + window: tauri::WebviewWindow, + state: tauri::State<'_, SessionObservations>, +) -> Result { + authorize_window(window.label())?; + state.summary() +} + +#[tauri::command] +pub fn reset_session_observations( + window: tauri::WebviewWindow, + state: tauri::State<'_, SessionObservations>, +) -> Result<(), SafeError> { + authorize_window(window.label())?; + state.reset() +} + +#[tauri::command] +pub fn preview_session_diagnostics( + window: tauri::WebviewWindow, + state: tauri::State<'_, SessionObservations>, +) -> Result { + authorize_window(window.label())?; + state.diagnostic_text() +} fn manager<'a, R: tauri::Runtime>( window: &tauri::WebviewWindow, state: &'a HostState, diff --git a/desktop/src-tauri/src/connections/session.rs b/desktop/src-tauri/src/connections/session.rs index ca97ba304b..064a1b97df 100644 --- a/desktop/src-tauri/src/connections/session.rs +++ b/desktop/src-tauri/src/connections/session.rs @@ -18,6 +18,9 @@ use super::profiles::{ProfileInput, ProfileRepository, ProfileView}; use crate::{ error::SafeError, + session_observations::{ + ObservationOutcome, ObservationUnit, ObservedOperation, SessionObservations, + }, transport::{ApiFailure, ServerApi, wire::*}, }; use serde::{Deserialize, Serialize}; @@ -154,6 +157,7 @@ struct Inner { last_write: Option, } pub struct ConnectionManager { + observations: SessionObservations, inner: Mutex, changed: watch::Sender, scope_reads: watch::Sender, @@ -190,6 +194,7 @@ impl ConnectionManager { ) -> Self { let (changed, _) = watch::channel(0); Self { + observations: SessionObservations::default(), inner: Mutex::new(Inner { profiles, reports: BTreeMap::new(), @@ -208,6 +213,9 @@ impl ConnectionManager { fn lock(&self) -> Result, SafeError> { self.inner.lock().map_err(|_| SafeError::Storage) } + pub fn observations(&self) -> &SessionObservations { + &self.observations + } fn invalidate(&self, inner: &mut Inner) -> Result<(), SafeError> { inner.generation = inner.generation.checked_add(1).ok_or(SafeError::Storage)?; self.changed.send_replace(inner.generation); @@ -303,7 +311,12 @@ impl ConnectionManager { self.invalidate(&mut inner)?; } ( - Arc::new(inner.profiles.api(id)?), + Arc::new( + inner + .profiles + .api(id)? + .with_observations(self.observations.clone()), + ), profile, inner.generation, self.changed.subscribe(), @@ -459,65 +472,77 @@ impl ConnectionManager { query: &str, cursor: Option<&str>, ) -> Result { - let ReadSnapshot { - api, - identity, - mut cancelled, - } = self.snapshot(generation, "list_scopes")?; - let mut query_cancelled = self.begin_scope_query(generation)?; - Self::cancellable( - &mut cancelled, - Self::cancellable(&mut query_cancelled, async { - self.identity_unchanged(&api, &identity, generation).await?; - api.scopes(query, cursor).await - }), - ) - .await + self.observations + .user_operation(ObservedOperation::ListScopes, async { + let ReadSnapshot { + api, + identity, + mut cancelled, + } = self.snapshot(generation, "list_scopes")?; + let mut query_cancelled = self.begin_scope_query(generation)?; + Self::cancellable( + &mut cancelled, + Self::cancellable(&mut query_cancelled, async { + self.identity_unchanged(&api, &identity, generation).await?; + api.scopes(query, cursor).await + }), + ) + .await + }) + .await } pub async fn default_scope(&self, generation: u32) -> Result { - let ReadSnapshot { - api, - identity, - mut cancelled, - } = self.snapshot(generation, "get_default_scope")?; - let mut query_cancelled = self.begin_scope_query(generation)?; - Self::cancellable( - &mut cancelled, - Self::cancellable(&mut query_cancelled, async { - self.identity_unchanged(&api, &identity, generation).await?; - api.default_scope().await - }), - ) - .await + self.observations + .user_operation(ObservedOperation::GetDefaultScope, async { + let ReadSnapshot { + api, + identity, + mut cancelled, + } = self.snapshot(generation, "get_default_scope")?; + let mut query_cancelled = self.begin_scope_query(generation)?; + Self::cancellable( + &mut cancelled, + Self::cancellable(&mut query_cancelled, async { + self.identity_unchanged(&api, &identity, generation).await?; + api.default_scope().await + }), + ) + .await + }) + .await } pub async fn select_scope( &self, generation: u32, id: &str, ) -> Result { - let ReadSnapshot { - api, - identity, - mut cancelled, - } = self.snapshot(generation, "get_scope")?; - let scope = Self::cancellable(&mut cancelled, async { - self.identity_unchanged(&api, &identity, generation).await?; - api.scope(id).await - }) - .await?; - let mut inner = self.lock()?; - if generation != inner.generation { - return Err(SafeError::StaleContext.into()); - } - self.invalidate(&mut inner)?; - inner - .active - .as_mut() - .ok_or(SafeError::NotConnected)? - .view - .scope = Some(scope); - drop(inner); - self.state().map_err(Into::into) + self.observations + .user_operation(ObservedOperation::GetScope, async { + let ReadSnapshot { + api, + identity, + mut cancelled, + } = self.snapshot(generation, "get_scope")?; + let scope = Self::cancellable(&mut cancelled, async { + self.identity_unchanged(&api, &identity, generation).await?; + api.scope(id).await + }) + .await?; + let mut inner = self.lock()?; + if generation != inner.generation { + return Err(SafeError::StaleContext.into()); + } + self.invalidate(&mut inner)?; + inner + .active + .as_mut() + .ok_or(SafeError::NotConnected)? + .view + .scope = Some(scope); + drop(inner); + self.state().map_err(Into::into) + }) + .await } fn memory_snapshot( &self, @@ -574,124 +599,152 @@ impl ConnectionManager { generation: u32, query: &str, ) -> Result { - let ( - ReadSnapshot { - api, - identity, - mut cancelled, - }, - context, - ) = self.memory_snapshot(generation, "search_memory")?; - let mut query_cancelled = self.begin_memory_read(generation)?; - Self::cancellable( - &mut cancelled, - Self::cancellable(&mut query_cancelled, async { - self.identity_unchanged(&api, &identity, generation).await?; - api.search(&context.scope_id, query).await - }), - ) - .await + self.observations + .user_operation(ObservedOperation::SearchMemory, async { + let ( + ReadSnapshot { + api, + identity, + mut cancelled, + }, + context, + ) = self.memory_snapshot(generation, "search_memory")?; + let mut query_cancelled = self.begin_memory_read(generation)?; + Self::cancellable( + &mut cancelled, + Self::cancellable(&mut query_cancelled, async { + self.identity_unchanged(&api, &identity, generation).await?; + api.search(&context.scope_id, query).await + }), + ) + .await + }) + .await } pub async fn atomic_memory_entry( &self, generation: u32, artifact: &ArtifactReference, ) -> Result { - let ( - ReadSnapshot { - api, - identity, - mut cancelled, - }, - context, - ) = self.memory_snapshot(generation, "get_artifact_revision")?; - let mut query_cancelled = self.begin_memory_read(generation)?; - Self::cancellable( - &mut cancelled, - Self::cancellable(&mut query_cancelled, async { - self.identity_unchanged(&api, &identity, generation).await?; - api.atomic_entry(&context.scope_id, artifact).await - }), - ) - .await + self.observations + .user_operation(ObservedOperation::GetArtifactRevision, async { + let ( + ReadSnapshot { + api, + identity, + mut cancelled, + }, + context, + ) = self.memory_snapshot(generation, "get_artifact_revision")?; + let mut query_cancelled = self.begin_memory_read(generation)?; + Self::cancellable( + &mut cancelled, + Self::cancellable(&mut query_cancelled, async { + self.identity_unchanged(&api, &identity, generation).await?; + api.atomic_entry(&context.scope_id, artifact).await + }), + ) + .await + }) + .await } pub async fn remember(&self, generation: u32, text: &str) -> Result { - let _permit = self.writes.try_acquire().map_err(|_| SafeError::Busy)?; - crate::transport::validate_text(text)?; - let ( - ReadSnapshot { - api, - identity, - mut cancelled, - }, - context, - ) = self.memory_snapshot(generation, "remember_memory")?; - Self::cancellable( - &mut cancelled, - self.identity_unchanged(&api, &identity, generation), - ) - .await?; - let mut record = WriteRecord { - operation_id: uuid::Uuid::new_v4().to_string(), - context, - status: WriteStatus::Pending, - references: vec![], - error: None, - }; - { - let mut inner = self.lock()?; - if inner.generation != generation { - return Err(SafeError::StaleContext.into()); + let mut observation = self.observations.begin( + ObservationUnit::UserOperation, + ObservedOperation::RememberMemory, + ); + let result = async { + let _permit = self.writes.try_acquire().map_err(|_| SafeError::Busy)?; + crate::transport::validate_text(text)?; + let ( + ReadSnapshot { + api, + identity, + mut cancelled, + }, + context, + ) = self.memory_snapshot(generation, "remember_memory")?; + Self::cancellable( + &mut cancelled, + self.identity_unchanged(&api, &identity, generation), + ) + .await?; + let mut record = WriteRecord { + operation_id: uuid::Uuid::new_v4().to_string(), + context, + status: WriteStatus::Pending, + references: vec![], + error: None, + }; + { + let mut inner = self.lock()?; + if inner.generation != generation { + return Err(SafeError::StaleContext.into()); + } + inner.last_write = Some(record.clone()); } + // No context cancellation after this point: the immutable API and Scope own the dispatched write. + let mut guard = WriteGuard { + manager: self, + armed: true, + }; + observation.write_dispatched(); + let response = api.remember(&record.context.scope_id, text).await; + let result = match response { + Ok(value) => { + record.status = WriteStatus::Succeeded; + record.references = value + .records + .iter() + .map(|record| record.artifact.clone()) + .collect(); + Some(value) + } + Err(error) => { + record.status = if error.dispatched + && !matches!( + error.code, + SafeError::Unauthorized + | SafeError::Forbidden + | SafeError::InvalidInput + | SafeError::NotFound + | SafeError::Conflict + | SafeError::AuthenticationUnavailable + | SafeError::RuntimeNotReady + | SafeError::Redirect + ) { + WriteStatus::Unknown + } else { + WriteStatus::Failed + }; + record.error = Some(error); + None + } + }; + let mut inner = self.lock()?; inner.last_write = Some(record.clone()); - } - // No context cancellation after this point: the immutable API and Scope own the dispatched write. - let mut guard = WriteGuard { - manager: self, - armed: true, - }; - let response = api.remember(&record.context.scope_id, text).await; - let result = match response { - Ok(value) => { - record.status = WriteStatus::Succeeded; - record.references = value - .records - .iter() - .map(|record| record.artifact.clone()) - .collect(); - Some(value) - } - Err(error) => { - record.status = if error.dispatched - && !matches!( - error.code, - SafeError::Unauthorized - | SafeError::Forbidden - | SafeError::InvalidInput - | SafeError::NotFound - | SafeError::Conflict - | SafeError::AuthenticationUnavailable - | SafeError::RuntimeNotReady - | SafeError::Redirect - ) { - WriteStatus::Unknown - } else { - WriteStatus::Failed - }; - record.error = Some(error); + guard.armed = false; + // A late response may report its original target, but may not disclose old body text in a new context. + let result = if inner.generation == generation { + result + } else { None - } - }; - let mut inner = self.lock()?; - inner.last_write = Some(record.clone()); - guard.armed = false; - // A late response may report its original target, but may not disclose old body text in a new context. - let result = if inner.generation == generation { - result - } else { - None - }; - Ok(WriteOutcome { record, result }) + }; + Ok(WriteOutcome { record, result }) + } + .await; + match &result { + Ok(outcome) => observation.write_finished( + match outcome.record.status { + WriteStatus::Succeeded => ObservationOutcome::Success, + WriteStatus::Failed => ObservationOutcome::Failure, + WriteStatus::Unknown | WriteStatus::Pending => ObservationOutcome::Unknown, + }, + outcome.record.error.as_ref().map(|error| error.code), + ), + Err(_) => observation.finish(&result), + } + result } async fn cancellable( cancelled: &mut watch::Receiver, diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 6c7726e1a3..787f646619 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -25,6 +25,7 @@ mod diagnostic_process; pub mod diagnostics; pub mod error; pub mod ipc; +pub mod session_observations; pub mod transport; use tauri::Manager; @@ -55,7 +56,10 @@ pub fn run() { commands::list_scopes, commands::cancel_scope_reads, commands::default_scope, - commands::select_scope + commands::select_scope, + commands::session_summary, + commands::reset_session_observations, + commands::preview_session_diagnostics ]) .setup(|app| { let manager = app @@ -76,6 +80,11 @@ pub fn run() { std::io::Error::other("Desktop profile storage is already in use").into(), ); } + let observations = manager + .as_ref() + .map(|manager| manager.observations().clone()) + .unwrap_or_default(); + app.manage(observations); app.manage(commands::HostState { manager }); app.manage(diagnostics::DiagnosticHost::new( app.path().app_data_dir().ok(), diff --git a/desktop/src-tauri/src/session_observations.rs b/desktop/src-tauri/src/session_observations.rs new file mode 100644 index 0000000000..810a486e62 --- /dev/null +++ b/desktop/src-tauri/src/session_observations.rs @@ -0,0 +1,477 @@ +/* + * Copyright (c) 2026 OceanBase. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//! Content-free, process-local observations. These are not a persistence or audit ledger. +use crate::{error::SafeError, transport::ApiFailure}; +use serde::Serialize; +use std::{ + collections::{BTreeMap, VecDeque}, + future::Future, + sync::{ + Arc, Mutex, TryLockError, + atomic::{AtomicU64, Ordering}, + }, + time::{Instant, SystemTime, UNIX_EPOCH}, +}; +use ts_rs::TS; + +pub const SESSION_OBSERVATION_LIMIT: usize = 512; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, TS)] +#[serde(rename_all = "snake_case")] +pub enum ObservationUnit { + HttpRequest, + UserOperation, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, TS)] +#[serde(rename_all = "snake_case")] +pub enum ObservedOperation { + GetLiveness, + GetReadiness, + GetCapabilities, + GetAccessPrincipal, + ListScopes, + GetScope, + GetDefaultScope, + RememberMemory, + SearchMemory, + GetArtifactRevision, +} +impl ObservedOperation { + pub(crate) fn id(self) -> &'static str { + match self { + Self::GetLiveness => "get_liveness", + Self::GetReadiness => "get_readiness", + Self::GetCapabilities => "get_capabilities", + Self::GetAccessPrincipal => "get_access_principal", + Self::ListScopes => "list_scopes", + Self::GetScope => "get_scope", + Self::GetDefaultScope => "get_default_scope", + Self::RememberMemory => "remember_memory", + Self::SearchMemory => "search_memory", + Self::GetArtifactRevision => "get_artifact_revision", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, TS)] +#[serde(rename_all = "snake_case")] +pub enum ObservationOutcome { + Success, + Failure, + Cancelled, + Unknown, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +pub struct SessionObservation { + pub unit: ObservationUnit, + pub operation: ObservedOperation, + pub outcome: ObservationOutcome, + pub completed_at_ms: u64, + pub duration_ms: f64, + pub http_status: Option, + pub error_code: Option, +} + +#[derive(Clone, Debug, Default, Serialize, TS)] +pub struct OutcomeCounts { + pub success: u32, + pub failure: u32, + pub cancelled: u32, + pub unknown: u32, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +pub struct OperationSummary { + pub unit: ObservationUnit, + pub operation: ObservedOperation, + pub outcomes: OutcomeCounts, + pub sample_count: u32, + pub min_duration_ms: f64, + pub mean_duration_ms: f64, + pub max_duration_ms: f64, +} + +#[derive(Serialize, TS)] +#[serde(rename_all = "camelCase")] +pub struct SessionCoverage { + pub desktop_session_only: bool, + pub persistence_evidence: bool, + pub http_operations: Vec, + pub user_operations: Vec, + pub identity_checks_included: bool, + pub installer_ops: bool, +} + +#[derive(Serialize, TS)] +#[serde(rename_all = "camelCase")] +pub struct SessionSummary { + pub schema_version: u8, + pub session_started_at_ms: u64, + pub window_started_at_ms: u64, + pub generated_at_ms: u64, + pub capacity: usize, + pub evicted: u64, + /// Process-lifetime loss count, deliberately not cleared by a window reset. + pub dropped_since_start: u64, + pub coverage: SessionCoverage, + pub operations: Vec, + pub recent: Vec, +} + +struct Window { + generation: u64, + started_at_ms: u64, + evicted: u64, + records: VecDeque, +} +struct Shared { + session_started_at_ms: u64, + generation: AtomicU64, + dropped: AtomicU64, + window: Mutex, +} + +#[derive(Clone)] +pub struct SessionObservations(Arc); +impl Default for SessionObservations { + fn default() -> Self { + let started_at_ms = now_ms(); + Self(Arc::new(Shared { + session_started_at_ms: started_at_ms, + generation: AtomicU64::new(0), + dropped: AtomicU64::new(0), + window: Mutex::new(Window { + generation: 0, + started_at_ms, + evicted: 0, + records: VecDeque::with_capacity(SESSION_OBSERVATION_LIMIT), + }), + })) + } +} +impl SessionObservations { + pub(crate) async fn user_operation( + &self, + operation: ObservedOperation, + work: impl Future>, + ) -> Result { + let mut observation = self.begin(ObservationUnit::UserOperation, operation); + let result = work.await; + observation.finish(&result); + result + } + + pub(crate) fn begin( + &self, + unit: ObservationUnit, + operation: ObservedOperation, + ) -> ObservationGuard { + ObservationGuard { + observations: self.clone(), + generation: self.0.generation.load(Ordering::Acquire), + started: Instant::now(), + unit, + operation, + outcome: ObservationOutcome::Cancelled, + error_code: None, + http_status: None, + } + } + + fn record(&self, generation: u64, observation: SessionObservation) { + // Observation contention/poisoning must never delay or fail the owning operation, including Drop. + match self.0.window.try_lock() { + Ok(mut window) if window.generation == generation => { + if window.records.len() == SESSION_OBSERVATION_LIMIT { + window.records.pop_front(); + window.evicted = window.evicted.saturating_add(1); + } + window.records.push_back(observation); + } + Ok(_) => (), // A pre-reset operation cannot repopulate the new window. + Err(_) => { + let _ = self + .0 + .dropped + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| { + Some(n.saturating_add(1)) + }); + } + } + } + + pub fn reset(&self) -> Result<(), SafeError> { + let mut window = self.0.window.try_lock().map_err(observation_lock_error)?; + window.generation = window.generation.checked_add(1).ok_or(SafeError::Storage)?; + window.started_at_ms = now_ms(); + window.records.clear(); + window.evicted = 0; + self.0 + .generation + .store(window.generation, Ordering::Release); + Ok(()) + } + + pub fn summary(&self) -> Result { + let window = self.0.window.try_lock().map_err(observation_lock_error)?; + let recent: Vec<_> = window.records.iter().cloned().collect(); + let started_at_ms = window.started_at_ms; + let evicted = window.evicted; + drop(window); + let mut groups = BTreeMap::<_, OperationSummary>::new(); + for record in &recent { + let group = groups + .entry((record.unit, record.operation)) + .or_insert_with(|| OperationSummary { + unit: record.unit, + operation: record.operation, + outcomes: OutcomeCounts::default(), + sample_count: 0, + min_duration_ms: record.duration_ms, + mean_duration_ms: 0.0, + max_duration_ms: record.duration_ms, + }); + match record.outcome { + ObservationOutcome::Success => group.outcomes.success += 1, + ObservationOutcome::Failure => group.outcomes.failure += 1, + ObservationOutcome::Cancelled => group.outcomes.cancelled += 1, + ObservationOutcome::Unknown => group.outcomes.unknown += 1, + } + group.sample_count += 1; + group.min_duration_ms = group.min_duration_ms.min(record.duration_ms); + group.max_duration_ms = group.max_duration_ms.max(record.duration_ms); + group.mean_duration_ms += + (record.duration_ms - group.mean_duration_ms) / f64::from(group.sample_count); + } + Ok(SessionSummary { + schema_version: 1, + session_started_at_ms: self.0.session_started_at_ms, + window_started_at_ms: started_at_ms, + generated_at_ms: now_ms(), + capacity: SESSION_OBSERVATION_LIMIT, + evicted, + dropped_since_start: self.0.dropped.load(Ordering::Relaxed), + coverage: SessionCoverage { + desktop_session_only: true, + persistence_evidence: false, + http_operations: vec![ + ObservedOperation::GetLiveness, + ObservedOperation::GetReadiness, + ObservedOperation::GetCapabilities, + ObservedOperation::GetAccessPrincipal, + ObservedOperation::ListScopes, + ObservedOperation::GetScope, + ObservedOperation::GetDefaultScope, + ObservedOperation::RememberMemory, + ObservedOperation::SearchMemory, + ObservedOperation::GetArtifactRevision, + ], + user_operations: vec![ + ObservedOperation::ListScopes, + ObservedOperation::GetScope, + ObservedOperation::GetDefaultScope, + ObservedOperation::RememberMemory, + ObservedOperation::SearchMemory, + ObservedOperation::GetArtifactRevision, + ], + identity_checks_included: true, + installer_ops: false, + }, + operations: groups.into_values().collect(), + recent, + }) + } + + /// Serialize only the native allowlisted projection, never connection or business state. + pub fn diagnostic_text(&self) -> Result { + serde_json::to_string_pretty(&self.summary()?).map_err(|_| SafeError::InvalidResponse) + } +} + +pub(crate) struct ObservationGuard { + observations: SessionObservations, + generation: u64, + started: Instant, + unit: ObservationUnit, + operation: ObservedOperation, + outcome: ObservationOutcome, + error_code: Option, + http_status: Option, +} +impl ObservationGuard { + pub(crate) fn http_status(&mut self, status: u16) { + self.http_status = Some(status); + } + + pub(crate) fn finish(&mut self, result: &Result) { + match result { + Ok(_) => self.outcome = ObservationOutcome::Success, + Err(error) => { + self.outcome = if self.outcome == ObservationOutcome::Unknown { + // Only the owning write result can resolve a dispatched write's uncertainty. + ObservationOutcome::Unknown + } else if error.code == SafeError::StaleContext { + ObservationOutcome::Cancelled + } else { + ObservationOutcome::Failure + }; + self.error_code = Some(error.code); + } + } + // Readiness 503 can be a successfully decoded fact, but remains an unsuccessful HTTP response. + if self.unit == ObservationUnit::HttpRequest + && self.http_status == Some(503) + && result.is_ok() + { + self.outcome = ObservationOutcome::Failure; + self.error_code = Some(SafeError::RuntimeNotReady); + } + } + + pub(crate) fn write_dispatched(&mut self) { + self.outcome = ObservationOutcome::Unknown; + } + + pub(crate) fn write_finished( + &mut self, + outcome: ObservationOutcome, + error_code: Option, + ) { + self.outcome = outcome; + self.error_code = error_code; + } +} +impl Drop for ObservationGuard { + fn drop(&mut self) { + self.observations.record( + self.generation, + SessionObservation { + unit: self.unit, + operation: self.operation, + outcome: self.outcome, + completed_at_ms: now_ms(), + duration_ms: self.started.elapsed().as_secs_f64() * 1000.0, + http_status: self.http_status, + error_code: self.error_code, + }, + ); + } +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) +} +fn observation_lock_error(error: TryLockError) -> SafeError { + match error { + TryLockError::WouldBlock => SafeError::Busy, + TryLockError::Poisoned(_) => SafeError::Storage, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::{Endpoint, ServerApi}; + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + sync::Notify, + }; + + #[tokio::test] + async fn collector_contention_cannot_fail_an_http_request_and_reports_lost_coverage() { + let observations = SessionObservations::default(); + let held = observations.clone(); + let (ready, received) = std::sync::mpsc::channel(); + let (release, wait) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + let _lock = held.0.window.lock().unwrap(); + ready.send(()).unwrap(); + wait.recv().unwrap(); + }); + received.recv().unwrap(); + assert!(matches!(observations.summary(), Err(SafeError::Busy))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let api = ServerApi::new( + Endpoint::parse(&format!("http://{}", listener.local_addr().unwrap())).unwrap(), + None, + None, + ) + .unwrap() + .with_observations(observations.clone()); + let response = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut bytes = [0; 4096]; + let received = socket.read(&mut bytes).await.unwrap(); + assert!(received > 0); + socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 15\r\nConnection: close\r\n\r\n{\"status\":\"ok\"}").await.unwrap(); + }); + assert_eq!(api.live().await.unwrap().status, "ok"); + response.await.unwrap(); + release.send(()).unwrap(); + worker.join().unwrap(); + let summary = observations.summary().unwrap(); + assert!(summary.recent.is_empty()); + assert_eq!(summary.dropped_since_start, 1); + observations.reset().unwrap(); + assert_eq!(observations.summary().unwrap().dropped_since_start, 1); + } + + #[tokio::test] + async fn poisoned_collector_cannot_suppress_task_cancellation() { + let observations = SessionObservations::default(); + let poisoned = observations.clone(); + let _ = std::thread::spawn(move || { + let _lock = poisoned.0.window.lock().unwrap(); + panic!("synthetic collector failure"); + }) + .join(); + assert!(matches!(observations.summary(), Err(SafeError::Storage))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let api = ServerApi::new( + Endpoint::parse(&format!("http://{}", listener.local_addr().unwrap())).unwrap(), + None, + None, + ) + .unwrap() + .with_observations(observations.clone()); + let accepted = Arc::new(Notify::new()); + let notify = accepted.clone(); + let response = tokio::spawn(async move { + let (_socket, _) = listener.accept().await.unwrap(); + notify.notify_one(); + std::future::pending::<()>().await; + }); + let request = tokio::spawn(async move { api.live().await }); + tokio::time::timeout(std::time::Duration::from_secs(3), accepted.notified()) + .await + .unwrap(); + request.abort(); + assert!(request.await.err().unwrap().is_cancelled()); + response.abort(); + assert_eq!(observations.0.dropped.load(Ordering::Relaxed), 1); + } +} diff --git a/desktop/src-tauri/src/transport/api.rs b/desktop/src-tauri/src/transport/api.rs index c48e684176..361fdc97a1 100644 --- a/desktop/src-tauri/src/transport/api.rs +++ b/desktop/src-tauri/src/transport/api.rs @@ -16,7 +16,11 @@ //! Operation-specific adapters. No renderer-supplied route, method or header is accepted. use super::{Endpoint, MAX_RESPONSE_BYTES, Transport, safe_network_error, wire::*}; -use crate::{credentials::Secret, error::SafeError}; +use crate::{ + credentials::Secret, + error::SafeError, + session_observations::{ObservationUnit, ObservedOperation, SessionObservations}, +}; use reqwest::{ Method, header::{AUTHORIZATION, HeaderValue}, @@ -75,6 +79,7 @@ pub struct ServerApi { transport: Transport, endpoint: Endpoint, credential: Option, + observations: SessionObservations, } impl ServerApi { pub fn new( @@ -86,11 +91,16 @@ impl ServerApi { transport: Transport::new(ca_pem)?, endpoint, credential, + observations: SessionObservations::default(), }) } + pub fn with_observations(mut self, observations: SessionObservations) -> Self { + self.observations = observations; + self + } async fn execute( &self, - operation: &str, + operation: ObservedOperation, scope: Option<&str>, query: &[(String, String)], body: Option, @@ -101,7 +111,7 @@ impl ServerApi { } async fn execute_path( &self, - operation: &str, + operation: ObservedOperation, scope: Option<&str>, query: &[(String, String)], body: Option, @@ -115,15 +125,15 @@ impl ServerApi { .map_err(|_| ApiFailure::before(SafeError::Busy))?; let manifest: serde_json::Value = serde_json::from_str(include_str!("operations.json")) .map_err(|_| ApiFailure::before(SafeError::InvalidResponse))?; - let descriptor = &manifest["operations"][operation]; + let descriptor = &manifest["operations"][operation.id()]; let method = descriptor["method"] .as_str() .ok_or(SafeError::InvalidResponse)?; let method = Method::from_bytes(method.as_bytes()).map_err(|_| SafeError::InvalidResponse)?; - let mut url = self.endpoint.operation_url(operation)?; + let mut url = self.endpoint.operation_url(operation.id())?; if let Some(artifact) = artifact { - if operation != "get_artifact_revision" + if operation != ObservedOperation::GetArtifactRevision || !descriptor["path"].as_str().is_some_and(|p| { p.ends_with( "/scopes/{scope_id}/artifacts/{family}/{artifact_id}/revisions/{revision}", @@ -177,76 +187,95 @@ impl ServerApi { request = request.json(&body); } let request = request.build().map_err(|_| SafeError::InvalidInput)?; - let failure = |code, request_id| ApiFailure { - code, - request_id, - dispatched: true, - }; - let mut response = self - .transport - .client - .execute(request) - .await - .map_err(|e| failure(safe_network_error(e), None))?; - let request_id = response - .headers() - .get("X-PowerContext-Request-ID") - .and_then(|v| v.to_str().ok()) - .filter(|v| { - !v.is_empty() - && v.len() <= 128 - && v.bytes() - .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') - }) - .map(str::to_owned); - let status = response.status().as_u16(); - if status != 200 && !(allow_not_ready && status == 503) { - let code = match status { - 300..=399 => SafeError::Redirect, - 401 => SafeError::Unauthorized, - 403 => SafeError::Forbidden, - 404 => SafeError::NotFound, - 409 => SafeError::Conflict, - 410 => SafeError::CursorExpired, - 400 | 422 => SafeError::InvalidInput, - 503 => server_error_code(&mut response).await, - 500..=599 => SafeError::Server, - _ => SafeError::InvalidResponse, + let mut observation = self + .observations + .begin(ObservationUnit::HttpRequest, operation); + let result = async { + let failure = |code, request_id| ApiFailure { + code, + request_id, + dispatched: true, }; - return Err(failure(code, request_id)); - } - if response - .content_length() - .is_some_and(|n| n > MAX_RESPONSE_BYTES as u64) - { - return Err(failure(SafeError::ResponseTooLarge, request_id)); - } - let mut bytes = Vec::new(); - while let Some(chunk) = response - .chunk() - .await - .map_err(|e| failure(safe_network_error(e), request_id.clone()))? - { - if bytes.len() + chunk.len() > MAX_RESPONSE_BYTES { + let mut response = self + .transport + .client + .execute(request) + .await + .map_err(|e| failure(safe_network_error(e), None))?; + let request_id = response + .headers() + .get("X-PowerContext-Request-ID") + .and_then(|v| v.to_str().ok()) + .filter(|v| { + !v.is_empty() + && v.len() <= 128 + && v.bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') + }) + .map(str::to_owned); + let status = response.status().as_u16(); + observation.http_status(status); + if status != 200 && !(allow_not_ready && status == 503) { + let code = match status { + 300..=399 => SafeError::Redirect, + 401 => SafeError::Unauthorized, + 403 => SafeError::Forbidden, + 404 => SafeError::NotFound, + 409 => SafeError::Conflict, + 410 => SafeError::CursorExpired, + 400 | 422 => SafeError::InvalidInput, + 503 => server_error_code(&mut response).await, + 500..=599 => SafeError::Server, + _ => SafeError::InvalidResponse, + }; + return Err(failure(code, request_id)); + } + if response + .content_length() + .is_some_and(|n| n > MAX_RESPONSE_BYTES as u64) + { return Err(failure(SafeError::ResponseTooLarge, request_id)); } - bytes.extend_from_slice(&chunk); + let mut bytes = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|e| failure(safe_network_error(e), request_id.clone()))? + { + if bytes.len() + chunk.len() > MAX_RESPONSE_BYTES { + return Err(failure(SafeError::ResponseTooLarge, request_id)); + } + bytes.extend_from_slice(&chunk); + } + serde_json::from_slice(&bytes) + .map_err(|_| failure(SafeError::InvalidResponse, request_id)) } - serde_json::from_slice(&bytes).map_err(|_| failure(SafeError::InvalidResponse, request_id)) + .await; + observation.finish(&result); + result } pub async fn live(&self) -> Result { - let result: HealthResponse = self.execute("get_liveness", None, &[], None, false).await?; + let result: HealthResponse = self + .execute(ObservedOperation::GetLiveness, None, &[], None, false) + .await?; if result.status != "ok" { return Err(SafeError::InvalidResponse.into()); } Ok(result) } pub async fn readiness(&self) -> Result { - self.execute("get_readiness", None, &[], None, true).await + self.execute(ObservedOperation::GetReadiness, None, &[], None, true) + .await } pub async fn principal(&self) -> Result { let result: AccessMeResponse = self - .execute("get_access_principal", None, &[], None, false) + .execute( + ObservedOperation::GetAccessPrincipal, + None, + &[], + None, + false, + ) .await?; if result.principal.id.is_empty() || result.principal.id.chars().count() > 255 @@ -260,7 +289,7 @@ impl ServerApi { Ok(result) } pub async fn capabilities(&self) -> Result { - self.execute("get_capabilities", None, &[], None, false) + self.execute(ObservedOperation::GetCapabilities, None, &[], None, false) .await } pub async fn scopes( @@ -282,7 +311,7 @@ impl ServerApi { query.push(("cursor".into(), cursor.into())); } let result: ScopePage = self - .execute("list_scopes", None, &query, None, false) + .execute(ObservedOperation::ListScopes, None, &query, None, false) .await?; if result.items.len() > 50 || result @@ -300,7 +329,7 @@ impl ServerApi { } pub async fn scope(&self, id: &str) -> Result { let result: ScopeDescriptor = self - .execute("get_scope", Some(id), &[], None, false) + .execute(ObservedOperation::GetScope, Some(id), &[], None, false) .await?; if result.scope_id != id { return Err(SafeError::InvalidResponse.into()); @@ -308,7 +337,7 @@ impl ServerApi { Ok(result) } pub async fn default_scope(&self) -> Result { - self.execute("get_default_scope", None, &[], None, false) + self.execute(ObservedOperation::GetDefaultScope, None, &[], None, false) .await } pub async fn remember( @@ -328,7 +357,13 @@ impl ServerApi { let body = serde_json::to_value(request) .map_err(|_| ApiFailure::before(SafeError::InvalidInput))?; let result: MemoryMutationResponse = self - .execute("remember_memory", None, &[], Some(body), false) + .execute( + ObservedOperation::RememberMemory, + None, + &[], + Some(body), + false, + ) .await?; if result .records @@ -360,7 +395,13 @@ impl ServerApi { let body = serde_json::to_value(request) .map_err(|_| ApiFailure::before(SafeError::InvalidInput))?; let result: SearchMemoryResponse = self - .execute("search_memory", None, &[], Some(body), false) + .execute( + ObservedOperation::SearchMemory, + None, + &[], + Some(body), + false, + ) .await?; if result.mode != AtomicMemorySearchMode::Text || result.hits.len() > 10 @@ -409,7 +450,7 @@ impl ServerApi { } let result: ArtifactRevision = self .execute_path( - "get_artifact_revision", + ObservedOperation::GetArtifactRevision, Some(scope), &[], None, diff --git a/desktop/src-tauri/tests/api.rs b/desktop/src-tauri/tests/api.rs index e2071ccbd8..5acf0d64e9 100644 --- a/desktop/src-tauri/tests/api.rs +++ b/desktop/src-tauri/tests/api.rs @@ -16,6 +16,9 @@ use powercontext_desktop::{ error::SafeError, + session_observations::{ + ObservationOutcome, ObservationUnit, ObservedOperation, SessionObservations, + }, transport::{ Endpoint, ServerApi, wire::{ArtifactReference, ReadinessStatus}, @@ -127,6 +130,74 @@ async fn readiness_failure_is_a_fact_and_write_failure_keeps_dispatch_uncertaint serde_json::json!({"scope_id":"selected-scope","kind":"note","text":"synthetic note"}) ); } + +#[tokio::test] +async fn request_observations_keep_http_status_and_exclude_private_response_and_request_data() { + let observations = SessionObservations::default(); + let (api, task) = fixture( + 503, + r#"{"status":"not_ready","checks":{"private-setting":"private-value"}}"#, + ) + .await; + let api = api.with_observations(observations.clone()); + assert_eq!( + api.readiness().await.unwrap().status, + ReadinessStatus::NotReady + ); + task.await.unwrap(); + let (api, task) = fixture(200, r#"{"status":"ok"}"#).await; + api.with_observations(observations.clone()) + .live() + .await + .unwrap(); + task.await.unwrap(); + let (api, task) = fixture(500, "private response and exception").await; + let error = api + .with_observations(observations.clone()) + .remember("private-scope", "private note") + .await + .err() + .unwrap(); + assert!(error.dispatched); + task.await.unwrap(); + let summary = observations.summary().unwrap(); + assert_eq!(summary.recent.len(), 3); + assert!( + summary + .recent + .iter() + .all(|record| record.unit == ObservationUnit::HttpRequest) + ); + assert_eq!(summary.recent[0].operation, ObservedOperation::GetReadiness); + assert_eq!(summary.recent[0].http_status, Some(503)); + assert_eq!(summary.recent[0].outcome, ObservationOutcome::Failure); + assert_eq!( + summary.recent[0].error_code, + Some(SafeError::RuntimeNotReady) + ); + assert_eq!(summary.recent[1].outcome, ObservationOutcome::Success); + assert_eq!(summary.recent[2].http_status, Some(500)); + assert_eq!(summary.recent[2].outcome, ObservationOutcome::Failure); + let text = observations.diagnostic_text().unwrap(); + for private in [ + "private", + "safe-request-1", + "127.0.0.1", + "proxy", + "中文", + "requestId", + "scopeId", + "endpoint", + ] { + assert!(!text.contains(private), "{private}"); + } + // Native preview is the same safe schema as the structured IPC result. + let projected: serde_json::Value = serde_json::from_str(&text).unwrap(); + assert_eq!( + projected["recent"], + serde_json::to_value(&summary.recent).unwrap() + ); +} #[tokio::test] async fn errors_are_distinct_without_anonymous_fallback() { for (status, code) in [ diff --git a/desktop/src-tauri/tests/ipc.rs b/desktop/src-tauri/tests/ipc.rs index f30bb54af8..85a56aa72e 100644 --- a/desktop/src-tauri/tests/ipc.rs +++ b/desktop/src-tauri/tests/ipc.rs @@ -226,3 +226,52 @@ fn memory_commands_reject_untrusted_callers_before_accessing_the_connection() { } } } + +#[test] +fn session_commands_work_without_a_connection_and_reject_untrusted_callers() { + use powercontext_desktop::{commands, session_observations::SessionObservations}; + use tauri::Manager; + let app = mock_builder() + .manage(SessionObservations::default()) + .invoke_handler(tauri::generate_handler![ + commands::session_summary, + commands::reset_session_observations, + commands::preview_session_diagnostics + ]) + .build(test_context()) + .unwrap(); + for command in [ + "session_summary", + "reset_session_observations", + "preview_session_diagnostics", + ] { + for (label, origin, allowed) in [ + ("main", LOCAL_ORIGIN, true), + ("other", LOCAL_ORIGIN, false), + ("main", "https://evil.example", false), + ] { + let window = app.get_webview_window(label).unwrap_or_else(|| { + tauri::WebviewWindowBuilder::new(&app, label, Default::default()) + .build() + .unwrap() + }); + let result = get_ipc_response( + &window, + InvokeRequest { + cmd: command.into(), + callback: CallbackFn(0), + error: CallbackFn(1), + url: origin.parse().unwrap(), + body: InvokeBody::default(), + headers: Default::default(), + invoke_key: INVOKE_KEY.into(), + }, + ); + assert_eq!( + result.is_ok(), + allowed, + "{command} {label} {origin}: {result:?}" + ); + } + } +} diff --git a/desktop/src-tauri/tests/memory.rs b/desktop/src-tauri/tests/memory.rs index 84c1d78140..70ff311c47 100644 --- a/desktop/src-tauri/tests/memory.rs +++ b/desktop/src-tauri/tests/memory.rs @@ -23,6 +23,9 @@ use powercontext_desktop::{ }, credentials::NativeVault, error::SafeError, + session_observations::{ + ObservationOutcome, ObservationUnit, ObservedOperation, SessionObservation, SessionSummary, + }, }; use std::sync::{ Arc, Mutex, @@ -221,6 +224,7 @@ async fn dispatched_write_keeps_its_original_target_and_rejects_duplicate_clicks #[tokio::test] async fn dispatched_server_failure_is_unknown_and_is_never_replayed() { let fixture = setup().await; + fixture.manager.observations().reset().unwrap(); fixture.fail.store(500, Ordering::Relaxed); fixture.release.notify_one(); let outcome = fixture @@ -233,6 +237,40 @@ async fn dispatched_server_failure_is_unknown_and_is_never_replayed() { assert_eq!(fixture.requests.lock().unwrap().len(), 1); let stored = fixture.manager.state().unwrap().last_write.unwrap(); assert_eq!(stored.status, WriteStatus::Unknown); + let summary = fixture.manager.observations().summary().unwrap(); + assert_eq!( + memory_observation(&summary, ObservationUnit::HttpRequest).outcome, + ObservationOutcome::Failure + ); + assert_eq!( + memory_observation(&summary, ObservationUnit::UserOperation).outcome, + ObservationOutcome::Unknown + ); + assert_eq!( + summary + .recent + .iter() + .filter(|r| r.unit == ObservationUnit::UserOperation) + .count(), + 1 + ); + assert!( + summary + .recent + .iter() + .any(|r| r.operation == ObservedOperation::GetAccessPrincipal + && r.unit == ObservationUnit::HttpRequest) + ); + let text = fixture.manager.observations().diagnostic_text().unwrap(); + for private in [ + "test note", + "scope-a", + &fixture.a, + "127.0.0.1", + "must not leak", + ] { + assert!(!text.contains(private), "{private}"); + } assert!( !serde_json::to_string(&stored) .unwrap() @@ -242,6 +280,7 @@ async fn dispatched_server_failure_is_unknown_and_is_never_replayed() { #[tokio::test] async fn success_without_records_does_not_invent_a_reference() { let fixture = setup().await; + fixture.manager.observations().reset().unwrap(); fixture.release.notify_one(); let result = fixture .manager @@ -277,6 +316,11 @@ async fn atomic_write_retains_every_exact_reference_after_context_changes() { fixture.release.notify_one(); let outcome = write.await.unwrap().unwrap(); assert_eq!(outcome.record.status, WriteStatus::Succeeded); + let summary = fixture.manager.observations().summary().unwrap(); + assert_eq!( + memory_observation(&summary, ObservationUnit::UserOperation).outcome, + ObservationOutcome::Success + ); assert_eq!(outcome.record.context.connection_id, fixture.a); if change_context { assert!(outcome.result.is_none()); @@ -313,6 +357,7 @@ async fn atomic_write_retains_every_exact_reference_after_context_changes() { #[tokio::test] async fn dropping_a_dispatched_write_preserves_unknown_metadata_without_body() { let fixture = setup().await; + fixture.manager.observations().reset().unwrap(); let generation = fixture.manager.state().unwrap().generation; let manager = fixture.manager.clone(); let write = @@ -327,6 +372,15 @@ async fn dropping_a_dispatched_write_preserves_unknown_metadata_without_body() { assert!(write.await.err().unwrap().is_cancelled()); let record = fixture.manager.state().unwrap().last_write.unwrap(); assert_eq!(record.status, WriteStatus::Unknown); + let summary = fixture.manager.observations().summary().unwrap(); + assert_eq!( + memory_observation(&summary, ObservationUnit::HttpRequest).outcome, + ObservationOutcome::Cancelled + ); + assert_eq!( + memory_observation(&summary, ObservationUnit::UserOperation).outcome, + ObservationOutcome::Unknown + ); assert!( !serde_json::to_string(&record) .unwrap() @@ -360,6 +414,7 @@ async fn absent_scope_never_uses_the_server_default_for_writing() { async fn explicit_write_rejections_are_failures_without_replay() { for (status, code) in [(409, SafeError::Conflict), (422, SafeError::InvalidInput)] { let fixture = setup().await; + fixture.manager.observations().reset().unwrap(); fixture.fail.store(status, Ordering::Relaxed); fixture.release.notify_one(); let result = fixture @@ -371,6 +426,11 @@ async fn explicit_write_rejections_are_failures_without_replay() { .await .unwrap(); assert_eq!(result.record.status, WriteStatus::Failed); + let summary = fixture.manager.observations().summary().unwrap(); + assert_eq!( + memory_observation(&summary, ObservationUnit::UserOperation).outcome, + ObservationOutcome::Failure + ); assert_eq!(result.record.error.unwrap().code, code); assert!(result.result.is_none()); assert_eq!(fixture.requests.lock().unwrap().len(), 1); @@ -396,3 +456,120 @@ async fn accepted_write_with_lost_response_remains_unknown_without_replay() { assert_eq!(accepted.len(), 1); assert_eq!(accepted[0]["text"], "synthetic accepted"); } + +fn memory_observation(summary: &SessionSummary, unit: ObservationUnit) -> &SessionObservation { + let mut records = summary.recent.iter().filter(|record| { + record.unit == unit && record.operation == ObservedOperation::RememberMemory + }); + let record = records.next().expect("observed memory operation"); + assert!(records.next().is_none()); + record +} + +#[tokio::test] +async fn reset_excludes_in_flight_observations_without_cancelling_the_write() { + let fixture = setup().await; + let generation = fixture.manager.state().unwrap().generation; + let manager = fixture.manager.clone(); + let write = + tokio::spawn(async move { manager.remember(generation, "private in-flight note").await }); + tokio::time::timeout( + std::time::Duration::from_secs(3), + fixture.started.notified(), + ) + .await + .unwrap(); + fixture.manager.observations().reset().unwrap(); + fixture.release.notify_one(); + let outcome = write.await.unwrap().unwrap(); + assert_eq!(outcome.record.status, WriteStatus::Succeeded); + assert!( + fixture + .manager + .observations() + .summary() + .unwrap() + .recent + .is_empty() + ); + assert_eq!(fixture.requests.lock().unwrap().len(), 1); +} + +#[tokio::test] +async fn successful_http_with_an_invalid_write_receipt_keeps_the_user_outcome_unknown() { + let fixture = setup_response(serde_json::json!({"changed":true,"records":[{ + "artifact":{"family":"atomic-memory","artifact_id":"private-artifact","revision":0}, + "kind":"note","text":"private returned text","state":"active","state_version":0,"merged_into_id":null + }]})).await; + fixture.manager.observations().reset().unwrap(); + fixture.release.notify_one(); + let outcome = fixture + .manager + .remember(fixture.manager.state().unwrap().generation, "private input") + .await + .unwrap(); + assert_eq!(outcome.record.status, WriteStatus::Unknown); + let summary = fixture.manager.observations().summary().unwrap(); + assert_eq!( + memory_observation(&summary, ObservationUnit::HttpRequest).outcome, + ObservationOutcome::Success + ); + assert_eq!( + memory_observation(&summary, ObservationUnit::UserOperation).outcome, + ObservationOutcome::Unknown + ); + assert!( + !fixture + .manager + .observations() + .diagnostic_text() + .unwrap() + .contains("private") + ); +} + +#[tokio::test] +async fn request_and_action_latency_use_milliseconds_with_distinct_sample_counts() { + let fixture = setup().await; + fixture.manager.observations().reset().unwrap(); + let generation = fixture.manager.state().unwrap().generation; + let manager = fixture.manager.clone(); + let started = std::time::Instant::now(); + let write = + tokio::spawn(async move { manager.remember(generation, "synthetic timing note").await }); + tokio::time::timeout( + std::time::Duration::from_secs(3), + fixture.started.notified(), + ) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + fixture.release.notify_one(); + assert_eq!( + write.await.unwrap().unwrap().record.status, + WriteStatus::Succeeded + ); + let elapsed_ms = started.elapsed().as_secs_f64() * 1000.0; + let summary = fixture.manager.observations().summary().unwrap(); + let request = memory_observation(&summary, ObservationUnit::HttpRequest); + let action = memory_observation(&summary, ObservationUnit::UserOperation); + assert!(request.duration_ms >= 20.0); + assert!(action.duration_ms >= request.duration_ms); + assert!(action.duration_ms <= elapsed_ms); + assert_eq!( + summary + .recent + .iter() + .filter(|r| r.unit == ObservationUnit::UserOperation) + .count(), + 1 + ); + assert_eq!( + summary + .recent + .iter() + .filter(|r| r.unit == ObservationUnit::HttpRequest) + .count(), + 3 + ); +} diff --git a/desktop/src-tauri/tests/observations.rs b/desktop/src-tauri/tests/observations.rs new file mode 100644 index 0000000000..0a69df2a2b --- /dev/null +++ b/desktop/src-tauri/tests/observations.rs @@ -0,0 +1,87 @@ +/* + * Copyright (c) 2026 OceanBase. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use powercontext_desktop::{ + connections::{profiles::ProfileRepository, session::ConnectionManager}, + credentials::NativeVault, + error::SafeError, + session_observations::{ + ObservationOutcome, ObservationUnit, ObservedOperation, SESSION_OBSERVATION_LIMIT, + SessionObservations, + }, +}; +use std::sync::Arc; + +#[tokio::test] +async fn retention_is_bounded_and_aggregates_only_the_retained_window() { + let dir = tempfile::tempdir().unwrap(); + let manager = ConnectionManager::new( + ProfileRepository::open( + dir.path().join("private-profile-path.json"), + Arc::new(NativeVault), + ) + .unwrap(), + ); + let observations = manager.observations(); + assert!(observations.summary().unwrap().recent.is_empty()); + manager.remember(0, "").await.err().unwrap(); + for _ in 0..SESSION_OBSERVATION_LIMIT + 6 { + assert_eq!( + manager + .search_memory(0, "private query") + .await + .err() + .unwrap() + .code, + SafeError::NotConnected + ); + } + let summary = observations.summary().unwrap(); + assert_eq!(summary.capacity, SESSION_OBSERVATION_LIMIT); + assert_eq!(summary.recent.len(), summary.capacity); + assert_eq!(summary.evicted, 7); + assert_eq!(summary.operations.len(), 1); + let group = &summary.operations[0]; + assert_eq!(group.unit, ObservationUnit::UserOperation); + assert_eq!(group.operation, ObservedOperation::SearchMemory); + assert_eq!(group.outcomes.failure, SESSION_OBSERVATION_LIMIT as u32); + assert_eq!(group.sample_count, SESSION_OBSERVATION_LIMIT as u32); + assert!(group.min_duration_ms >= 0.0); + assert!(group.mean_duration_ms >= group.min_duration_ms); + assert!(group.mean_duration_ms <= group.max_duration_ms); + assert!(summary.recent.iter().all( + |record| record.outcome == ObservationOutcome::Failure && record.http_status.is_none() + )); + let text = observations.diagnostic_text().unwrap(); + assert!(!text.contains("private query")); + assert!(!text.contains("private-profile-path")); + assert!(!summary.coverage.installer_ops); + assert!(summary.coverage.identity_checks_included); + observations.reset().unwrap(); + let reset = observations.summary().unwrap(); + assert_eq!(reset.session_started_at_ms, summary.session_started_at_ms); + assert!(reset.window_started_at_ms >= summary.window_started_at_ms); + assert_eq!(reset.evicted, 0); + assert!(reset.recent.is_empty()); + assert!(reset.operations.is_empty()); + assert!( + SessionObservations::default() + .summary() + .unwrap() + .recent + .is_empty() + ); +} diff --git a/desktop/src-tauri/tests/session.rs b/desktop/src-tauri/tests/session.rs index b7f5a5babc..c686e84476 100644 --- a/desktop/src-tauri/tests/session.rs +++ b/desktop/src-tauri/tests/session.rs @@ -223,6 +223,22 @@ async fn switching_connection_cancels_a_slow_scope_read_before_it_can_return_old .unwrap(); assert_eq!(result.err().unwrap().code, SafeError::StaleContext); assert_eq!(manager.state().unwrap().active.unwrap().connection_id, b); + let summary = manager.observations().summary().unwrap(); + for unit in [ + powercontext_desktop::session_observations::ObservationUnit::HttpRequest, + powercontext_desktop::session_observations::ObservationUnit::UserOperation, + ] { + assert!(summary.recent.iter().any(|record| record.unit == unit + && record.operation + == powercontext_desktop::session_observations::ObservedOperation::ListScopes + && record.outcome + == powercontext_desktop::session_observations::ObservationOutcome::Cancelled)); + } + let text = manager.observations().diagnostic_text().unwrap(); + assert!(!text.contains(&a)); + assert!(!text.contains(&b)); + assert!(!text.contains(&server.endpoint)); + server.release.notify_one(); } #[tokio::test] diff --git a/desktop/ui/src/generated/ipc.ts b/desktop/ui/src/generated/ipc.ts index 7474c29357..e96086c617 100644 --- a/desktop/ui/src/generated/ipc.ts +++ b/desktop/ui/src/generated/ipc.ts @@ -46,6 +46,18 @@ export type MemorySaveResponse = { changed: boolean, records: Array, }; export type MemorySearchResponse = { hits: Array, }; export type AtomicMemoryDetail = { artifact: ArtifactReference, kind: string, text: string, source_refs: Array, }; +export type ObservationUnit = "http_request" | "user_operation"; +export type ObservedOperation = "get_liveness" | "get_readiness" | "get_capabilities" | "get_access_principal" | "list_scopes" | "get_scope" | "get_default_scope" | "remember_memory" | "search_memory" | "get_artifact_revision"; +export type ObservationOutcome = "success" | "failure" | "cancelled" | "unknown"; +export type SessionObservation = { unit: ObservationUnit, operation: ObservedOperation, outcome: ObservationOutcome, completedAtMs: number, durationMs: number, httpStatus: number | null, errorCode: SafeError | null, }; +export type OutcomeCounts = { success: number, failure: number, cancelled: number, unknown: number, }; +export type OperationSummary = { unit: ObservationUnit, operation: ObservedOperation, outcomes: OutcomeCounts, sampleCount: number, minDurationMs: number, meanDurationMs: number, maxDurationMs: number, }; +export type SessionCoverage = { desktopSessionOnly: boolean, persistenceEvidence: boolean, httpOperations: Array, userOperations: Array, identityChecksIncluded: boolean, installerOps: boolean, }; +export type SessionSummary = { schemaVersion: number, sessionStartedAtMs: number, windowStartedAtMs: number, generatedAtMs: number, capacity: number, evicted: number, +/** + * Process-lifetime loss count, deliberately not cleared by a window reset. + */ +droppedSinceStart: number, coverage: SessionCoverage, operations: Array, recent: Array, }; export type AccessAction = "server.observe" | "server.admin" | "scope.read" | "scope.contribute" | "scope.review" | "scope.delegate" | "scope.admin" | "artifact.read" | "artifact.write" | "artifact.share" | "handoff.evidence.inspect" | "handoff.acknowledge" | "prompt.use"; export type AccessControlMode = "disabled" | "enforced"; export type AccessMeResponse = { principal: AccessPrincipal, mode: AccessControlMode, resource_kinds: Array, provider_capabilities: AccessProviderCapabilities, artifact_families: Array, }; diff --git a/desktop/ui/src/shared/ipc.ts b/desktop/ui/src/shared/ipc.ts index 65d2b1089b..b9e75642d5 100644 --- a/desktop/ui/src/shared/ipc.ts +++ b/desktop/ui/src/shared/ipc.ts @@ -26,6 +26,7 @@ import type { DesktopState, ProfileInput, ScopePage, + SessionSummary, } from "../generated/ipc"; export async function getFoundationInfo(): Promise { if (!isTauri()) return null; @@ -33,6 +34,10 @@ export async function getFoundationInfo(): Promise { } export const desktopApi = { + sessionSummary: () => invoke("session_summary"), + resetSessionObservations: () => invoke("reset_session_observations"), + previewSessionDiagnostics: () => + invoke("preview_session_diagnostics"), remember: (generation: number, text: string) => invoke("remember_memory", { generation, text }), search: (generation: number, query: string) => diff --git a/docs/en/development/desktop-session-observations.md b/docs/en/development/desktop-session-observations.md new file mode 100644 index 0000000000..d865553951 --- /dev/null +++ b/docs/en/development/desktop-session-observations.md @@ -0,0 +1,106 @@ +--- +title: Desktop session observations +description: Native request and user-operation measurements, bounded retention, and safe diagnostic projection. +--- + +# Desktop session observations + +Desktop owns an in-memory observation window under [#2107](https://github.com/oceanbase/powercontext/issues/2107). +Collection starts with the native host and does not depend on opening Observation. The typed native commands +`session_summary`, `reset_session_observations`, and `preview_session_diagnostics` require the trusted main window +and its capability. They remain available without an active connection or usable profile storage. They do not issue +HTTP requests, probe readiness, launch diagnostics, write the clipboard, or upload anything. The This session page +and explicit preview/copy interaction belong to the separate UI delivery; the page remains unavailable until integrated. + +## Measurement units and coverage + +Every record declares `unit` and the owning operation ID in `operation`. + +| Unit | Counted work | Duration in milliseconds | +| --- | --- | --- | +| `http_request` | One request handed to the native HTTP client | Monotonic time from dispatch through bounded response reading and JSON decoding, or failure/cancellation | +| `user_operation` | One native Memory or Scope action invocation, including rejected invocations | Monotonic time from entering the native action through context/identity checks, response validation and result construction, or failure/cancellation | + +HTTP coverage includes `get_liveness`, `get_readiness`, `get_capabilities`, `get_access_principal`, `list_scopes`, +`get_scope`, `get_default_scope`, `remember_memory`, `search_memory`, and `get_artifact_revision`. Native connection +checks contribute their individual HTTP requests, not an additional user operation. Identity revalidation and its +existing anonymous-access readiness fallback also contribute HTTP requests. A user action can therefore issue multiple +HTTP requests. The summary's `coverage.httpOperations` and `coverage.identityChecksIncluded` disclose that coverage. + +User-operation coverage is the six Memory/Scope operations listed in `coverage.userOperations`: `list_scopes`, +`get_scope`, `get_default_scope`, `remember_memory`, `search_memory`, and `get_artifact_revision`. It does not include +profile editing, connection checks, renderer debounce, UI confirmation time, or clipboard actions. Native pre-dispatch +validation/concurrency refusals can produce a failed user operation without an HTTP record. HTTP latency is client +request time, not Server processing time; user-operation latency is native action time, not the entire UI journey. + +`coverage.desktopSessionOnly` is true and `coverage.persistenceEvidence` is false. This is this Desktop process's +observation, not global Server/Agent tracing or authoritative evidence of persistence. Installer/Ops and local CLI +diagnostics are not collected (`coverage.installerOps` is false). A future local-setup section must project the owning +Installer/Ops phase/result and supported opaque operation reference. It must not import stdout, paths, configuration, +or journal contents, create another journal, or include local lifecycle work in HTTP latency. + +## Outcomes + +`outcome` is `success`, `failure`, `cancelled`, or `unknown`. HTTP success means a 200 response was read and decoded; +operation-specific validation can still fail afterward. Readiness 503 remains an HTTP failure with +`errorCode: runtime_not_ready`, even when the adapter successfully returns a structured not-ready fact. A cancelled +HTTP future records `cancelled`; it does not imply that a remote write was cancelled. + +Read user operations use their native return result. `stale_context` is classified as cancellation, including an +invocation made with an obsolete context. Saving uses the existing `WriteRecord.status`, not whether the outer +`Result` is `Ok`. A successful save with zero records remains successful. Explicit write rejections remain failures; +dispatch followed by a network failure, 500, invalid receipt, or interruption can leave the user operation `unknown`. +A decoded HTTP 200 can coexist with an unknown save if its receipt fails operation-specific validation. Dropping a +save after entering the existing write dispatch boundary preserves unknown semantics. Observation never retargets, +replays, retries, cancels, or confirms a write. + +## Retention, reset and cost + +The shared window retains at most **512 terminal records**, across both units and all connections/Scopes, in native +completion order. Pending work is absent until completion or cancellation. Each insertion has constant bounded work; +an overflow evicts the oldest record. Summaries copy and aggregate at most 512 records, with groups bounded by the +declared unit/operation vocabulary. There is no history scan, persistence, background task, external exporter, or +per-context collection. Closing Desktop discards all observations; reopening creates an empty session. + +Reset clears retained records, groups and the eviction count and moves the window start. It does not reset the +Desktop session start, affect business state, or cancel work. Guards created before reset cannot repopulate the new +window. Requests started after reset can appear even if their enclosing user operation began earlier. + +Connection, Principal and Scope transitions keep anonymous process-wide observations. No record stores target +identity, and no historical or late result is assigned to a newly selected target. Existing private-context invalidation, +read cancellation and immutable dispatched-write targeting remain owned by the connection manager. + +Recording uses a nonblocking lock and no I/O. Contention or a poisoned collector drops the observation without +changing business results or suppressing cancellation. `droppedSinceStart` counts such losses over the process +lifetime and intentionally survives reset, so resets cannot conceal incomplete coverage. Snapshot/reset contention +returns `busy`; a poisoned collector returns `storage`, rather than fabricating an empty successful summary. + +## Safe projection contract + +`SessionSummary` is an explicit Rust projection; it never serializes `DesktopState`, `WriteRecord`, raw requests, +responses or exceptions. The native preview serializes only this projection as pretty JSON. Structured and text +outputs use the same allowlist; no renderer redaction is required for safety. + +| Fields | Meaning and units | +| --- | --- | +| `schemaVersion` | Safe-summary schema version, currently 1 | +| `sessionStartedAtMs`, `windowStartedAtMs`, `generatedAtMs` | UTC Unix epoch milliseconds: process session start, last reset/start, snapshot generation | +| `capacity`, `evicted`, `droppedSinceStart` | Maximum retained terminal records, evictions since reset, recording losses since process start | +| `coverage` | Declared native coverage and limits described above | +| `recent` | Retained records in completion order; not a complete session history after eviction or loss | +| `unit`, `operation`, `outcome` | Fixed measurement unit, allowlisted operation ID and result enum | +| `completedAtMs`, `durationMs` | UTC completion timestamp in epoch milliseconds and nonnegative monotonic elapsed milliseconds | +| `httpStatus`, `errorCode` | Observed HTTP status if received, and existing `SafeError` enum if applicable; otherwise null | +| `operations` | Groups by unit and operation, computed only from `recent` | +| `outcomes`, `sampleCount` | Retained terminal counts by outcome and total sample count | +| `minDurationMs`, `meanDurationMs`, `maxDurationMs` | Elapsed milliseconds across all retained outcomes in that group, including cancelled and unknown work | + +The collector accepts no free-form diagnostic fields. Memory/Source content, queries, prompts/responses, credentials, +raw endpoints, profile names/IDs, Principal/Scope/Artifact IDs, private paths, request headers, raw errors and installer +output never enter records. Server request IDs remain available through existing operation failures but are omitted +from this export. It creates no correlation IDs or execution-identity metric labels and configures no operational logs, +metrics, spans or telemetry backend. + +Native regression checks use synthetic loopback HTTP fixtures and mocked Tauri IPC. They protect bounded retention, +reset with in-flight writes, request/action counting, readiness and write outcomes, cancellation, sensitive-value +exclusion, permissions and collector failure isolation. They do not qualify a real Server or installed clipboard. diff --git a/docs/en/development/meta.json b/docs/en/development/meta.json index 39bb09b4cd..3e7288767e 100644 --- a/docs/en/development/meta.json +++ b/docs/en/development/meta.json @@ -11,6 +11,7 @@ "observability", "remote-access-implementation", "dashboard", - "desktop-delivery-plan" + "desktop-delivery-plan", + "desktop-session-observations" ] } diff --git a/docs/en/development/observability.md b/docs/en/development/observability.md index 869c242fcb..84a7c81dbf 100644 --- a/docs/en/development/observability.md +++ b/docs/en/development/observability.md @@ -9,6 +9,9 @@ Before adding a log, metric, or span, identify the question it answers and the b Use the [operational field contract](../docs/operate/observability.md) for names, units, outcomes, error categories, and correlation semantics. [RFC 0046](../rfcs/0046_observability_foundations.md) defines privacy and signal ownership. +For the Desktop host's process-local measurements and native safe export, use the separate +[Desktop session observation contract](desktop-session-observations.md). It does not configure Server logging or tracing. + ## Add an operational event Use a module logger and `powercontext._logging.log_safely`. Keep the message fixed and put approved diagnostic values diff --git a/docs/zh/development/desktop-session-observations.md b/docs/zh/development/desktop-session-observations.md new file mode 100644 index 0000000000..cfb1b7ac16 --- /dev/null +++ b/docs/zh/development/desktop-session-observations.md @@ -0,0 +1,92 @@ +--- +title: Desktop 会话观察 +description: 原生请求与用户操作的测量口径、有界保留和安全诊断投影。 +--- + +# Desktop 会话观察 + +Desktop 为 [#2107](https://github.com/oceanbase/powercontext/issues/2107) 提供内存中的观察窗口。 +采集随原生宿主启动，不依赖打开观察页面。原生命令 `session_summary`、`reset_session_observations` 和 +`preview_session_diagnostics` 只允许拥有对应 capability 的可信主窗口调用；没有活动连接或配置存储不可用时仍可调用。 +这些命令不会发起 HTTP 请求、探测就绪状态、运行本地诊断、写入剪贴板或上传数据。本次会话页面以及显式预览、复制交互 +由独立的 UI 工作交付；完成集成之前，页面继续显示不可用。 + +## 测量单位与覆盖范围 + +每条记录都通过 `unit` 和 `operation` 明确测量单位及所属操作 ID。 + +| 单位 | 计数对象 | 耗时口径，单位为毫秒 | +| --- | --- | --- | +| `http_request` | 一次交给原生 HTTP 客户端的请求 | 从派发到完成有界响应读取、JSON 解码，或失败、取消的单调时钟耗时 | +| `user_operation` | 一次原生 Memory 或 Scope 操作调用，包括被拒绝的调用 | 从进入原生操作到完成上下文与身份检查、响应校验及结果构造，或失败、取消的单调时钟耗时 | + +HTTP 覆盖 `get_liveness`、`get_readiness`、`get_capabilities`、`get_access_principal`、`list_scopes`、 +`get_scope`、`get_default_scope`、`remember_memory`、`search_memory` 和 `get_artifact_revision`。 +连接检查只计入它实际发出的各项 HTTP 请求，不额外产生用户操作。身份复核及现有匿名访问的就绪检查回退也计入 HTTP 请求。 +因此，一次用户操作可能产生多个请求；`coverage.httpOperations` 和 `coverage.identityChecksIncluded` 明确说明该范围。 + +用户操作覆盖 `coverage.userOperations` 中的六项 Memory/Scope 操作：`list_scopes`、`get_scope`、 +`get_default_scope`、`remember_memory`、`search_memory` 和 `get_artifact_revision`。不包含配置编辑、连接检查、 +渲染层防抖、界面确认等待或剪贴板操作。派发前的本地校验和并发限制可能产生失败的用户操作，而没有 HTTP 记录。 +HTTP 延迟是客户端请求耗时，不能解释为 Server 内部处理耗时；用户操作延迟是原生操作耗时，不代表完整界面旅程。 + +`coverage.desktopSessionOnly` 为 true，`coverage.persistenceEvidence` 为 false。这些数据只覆盖当前 Desktop 进程， +不是全局 Server/Agent 链路，也不是持久化成功的权威证据。不采集 Installer/Ops 或本地 CLI 诊断， +`coverage.installerOps` 为 false。后续本地安装诊断区只能投影所属 Installer/Ops 的安全阶段、结果和受支持的不透明操作引用。 +不得引入 stdout、路径、配置或 journal 内容，不得另建 journal，也不能将本地生命周期操作计入 HTTP 延迟。 + +## 结果语义 + +`outcome` 为 `success`、`failure`、`cancelled` 或 `unknown`。HTTP 成功表示已读取并解码 200 响应；后续操作级校验仍可能失败。 +即使适配器正常返回结构化的未就绪事实，readiness 的 503 仍计为 HTTP 失败，`errorCode` 为 `runtime_not_ready`。 +HTTP Future 被取消时记录 `cancelled`，不代表远端写入已经取消。 + +读取类用户操作采用其原生返回结果。`stale_context` 归为取消，包括携带过期上下文发起的调用。 +保存操作复用现有 `WriteRecord.status`，不根据外层 `Result` 是否为 `Ok` 判断成功。成功且返回零条记录仍为成功。 +明确拒绝的写入仍为失败；派发后的网络故障、500、无效回执或中断可能使用户操作保持 `unknown`。 +如果操作级回执校验失败，已正常解码的 HTTP 200 可以与未知的保存结果同时存在。 +在现有写入派发边界之后丢弃保存任务，继续保持未知结果。观察不会重定向、重放、重试、取消或确认业务写入。 + +## 保留、重置与成本 + +所有测量单位、连接和 Scope 共享一个窗口，按原生完成顺序最多保留 **512 条终态记录**。 +尚未结束的工作在完成或取消前不出现在记录中。每次插入的工作量有界且为常量；溢出时淘汰最早的记录。 +摘要最多复制和汇总 512 条记录，分组数量受固定的单位及操作词汇约束。没有历史扫描、持久化、后台任务、外部导出器或 +按上下文增长的集合。退出 Desktop 会丢弃全部观察；重新打开时建立空会话。 + +重置清空保留记录、分组和淘汰数量，更新窗口起点；不改变会话起点或业务状态，也不取消任务。 +重置前创建的观察守卫不会重新填入新窗口。重置后才开始的 HTTP 请求仍可能出现，即使它所属的用户操作开始于重置前。 + +连接、Principal 和 Scope 切换保留进程范围内的匿名观察。记录不存储目标身份，历史结果和晚到结果不会被归到新选择的目标。 +原有私有上下文失效、读取取消和已派发写入的不可变目标仍由连接管理器负责。 + +记录过程使用非阻塞锁，不执行 I/O。锁竞争或采集器锁中毒只会丢弃观察，不改变业务结果，也不抑制取消。 +`droppedSinceStart` 统计整个进程的此类丢失数量，重置后仍保留，避免重置掩盖覆盖缺失。 +读取或重置时遇到锁竞争返回 `busy`，采集器锁中毒返回 `storage`，不会伪造成功的空摘要。 + +## 安全投影契约 + +`SessionSummary` 是显式的 Rust 投影，不序列化 `DesktopState`、`WriteRecord`、原始请求、响应或异常。 +原生预览只将该投影格式化为 JSON。结构化输出和文本输出采用相同的白名单，安全性不依赖渲染层脱敏。 + +| 字段 | 含义和单位 | +| --- | --- | +| `schemaVersion` | 安全摘要模式版本，当前为 1 | +| `sessionStartedAtMs`、`windowStartedAtMs`、`generatedAtMs` | UTC Unix epoch 毫秒：进程会话起点、最近重置或启动时间、摘要生成时间 | +| `capacity`、`evicted`、`droppedSinceStart` | 终态记录上限、重置以来的淘汰数量、进程启动以来的采集丢失数量 | +| `coverage` | 上述原生采集范围和限制 | +| `recent` | 按完成顺序保留的记录；存在淘汰或丢失时，不代表完整会话历史 | +| `unit`、`operation`、`outcome` | 固定的测量单位、白名单操作 ID 和结果枚举 | +| `completedAtMs`、`durationMs` | UTC 完成时间的 epoch 毫秒，以及非负的单调时钟耗时，单位为毫秒 | +| `httpStatus`、`errorCode` | 已收到的 HTTP 状态、适用的现有 `SafeError` 枚举；无对应值时为 null | +| `operations` | 仅根据 `recent`，按单位和操作计算的分组 | +| `outcomes`、`sampleCount` | 各类终态的保留数量和样本总数 | +| `minDurationMs`、`meanDurationMs`、`maxDurationMs` | 分组内所有保留终态的最小、平均及最大耗时，单位为毫秒，包含已取消和未知结果 | + +采集器不接收自由格式诊断字段。Memory/Source 内容、查询、提示词与响应、凭据、原始 endpoint、配置名称与 ID、 +Principal/Scope/Artifact ID、私有路径、请求头、原始错误和安装输出都不进入记录。 +Server 请求 ID 仍可通过现有操作失败结果获取，但不进入会话导出。采集器不创建关联 ID 或执行身份指标标签， +也不配置运行日志、指标、span 或遥测后端。 + +原生回归检查使用模拟的 loopback HTTP 服务和 mock Tauri IPC，覆盖有界保留、在途写入期间重置、请求与操作计数、 +就绪及写入结果、取消、敏感值排除、权限和采集故障隔离。这些检查不构成真实 Server 或安装包剪贴板验收。 diff --git a/docs/zh/development/meta.json b/docs/zh/development/meta.json index 26d69af5ee..cf3a9517de 100644 --- a/docs/zh/development/meta.json +++ b/docs/zh/development/meta.json @@ -11,6 +11,7 @@ "observability", "remote-access-implementation", "dashboard", - "desktop-delivery-plan" + "desktop-delivery-plan", + "desktop-session-observations" ] } diff --git a/docs/zh/development/observability.md b/docs/zh/development/observability.md index 8a6e401917..7ae5e55ac0 100644 --- a/docs/zh/development/observability.md +++ b/docs/zh/development/observability.md @@ -10,6 +10,9 @@ description: 使用稳定字段、正确的 correlation 和 failure isolation [operational field contract](../docs/operate/observability.md)。 [RFC 0046](../rfcs/0046_observability_foundations.md) 定义 privacy 与 signal ownership。 +Desktop 宿主的进程内测量与原生安全导出遵循独立的 [Desktop 会话观察契约](desktop-session-observations.md)， +不会配置 Server 日志或 tracing。 + ## 新增 operational event 使用 module logger 和 `powercontext._logging.log_safely`。Message 使用固定文本，批准的诊断值放入 structured fields。 From 1c7f830ec42f6d2213b5e455601465e81166e23b Mon Sep 17 00:00:00 2001 From: sunlight798 <3281498087@qq.com> Date: Sun, 25 Oct 2026 16:57:21 +0800 Subject: [PATCH 2/2] test(desktop): stabilize session summary JSON comparison --- desktop/src-tauri/tests/api.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/desktop/src-tauri/tests/api.rs b/desktop/src-tauri/tests/api.rs index 5acf0d64e9..d3e54348f8 100644 --- a/desktop/src-tauri/tests/api.rs +++ b/desktop/src-tauri/tests/api.rs @@ -192,11 +192,11 @@ async fn request_observations_keep_http_status_and_exclude_private_response_and_ assert!(!text.contains(private), "{private}"); } // Native preview is the same safe schema as the structured IPC result. + // Round-trip the expected JSON too so float parsing has the same rounding behavior. let projected: serde_json::Value = serde_json::from_str(&text).unwrap(); - assert_eq!( - projected["recent"], - serde_json::to_value(&summary.recent).unwrap() - ); + let expected: serde_json::Value = + serde_json::from_str(&serde_json::to_string_pretty(&summary.recent).unwrap()).unwrap(); + assert_eq!(projected["recent"], expected); } #[tokio::test] async fn errors_are_distinct_without_anonymous_fallback() {