feat(connect): add consent-bound profile exports (#7716)

feat: add consent-bound profile exports
2026-09-23 19:06:30 +00:00 · 2026-09-13 04:33:30 +08:00
parent d8ef9e803e
commit 95e1b84ecf
12 changed files with 1882 additions and 5 deletions
@@ -127,6 +127,97 @@ pub enum ConnectCommands {
    Register(ConnectRegisterOpts),
    /// Import, verify, or inspect a signed Connect service license
    License(ConnectLicenseOpts),
+    /// Capture a consent-bound local profile and write a signed export
+    Profile(ConnectProfileOpts),
+}
+
+/// `connect profile` options.
+#[derive(Args, Clone)]
+pub struct ConnectProfileOpts {
+    /// Directory containing an enrolled Connect device identity
+    #[arg(long = "state-dir")]
+    pub state_dir: PathBuf,
+
+    /// New local archive path; an existing file is never replaced
+    #[arg(long)]
+    pub output: PathBuf,
+
+    /// Profile producer to run
+    #[arg(long, value_enum)]
+    pub tool: ConnectProfileTool,
+
+    /// Thread source required by the threads producer
+    #[arg(long = "thread-scope", value_enum)]
+    pub thread_scope: Option<ConnectThreadProfileScope>,
+
+    /// Negotiated producer schema version
+    #[arg(long = "schema-version", default_value_t = 1)]
+    pub schema_version: u16,
+
+    /// Negotiated producer capability, such as profile.memory@1
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub capability: String,
+
+    /// Organization resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub organization: String,
+
+    /// Cluster resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub cluster: String,
+
+    /// Cluster-device resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub device: String,
+
+    /// UUIDv7 diagnostic run identifier issued by Connect
+    #[arg(long = "run-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub run_uid: String,
+
+    /// UUIDv7 artifact identifier issued by Connect
+    #[arg(long = "artifact-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub artifact_uid: String,
+
+    /// UUIDv7 consent identifier issued by Connect
+    #[arg(long = "consent-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub consent_uid: String,
+
+    /// Consent policy revision bound to this capture
+    #[arg(long = "policy-revision")]
+    pub policy_revision: u64,
+
+    /// Consent expiry as UTC Unix seconds
+    #[arg(long = "consent-expires-at")]
+    pub consent_expires_at_unix: i64,
+
+    /// Artifact expiry as UTC Unix seconds
+    #[arg(long = "expires-at")]
+    pub expires_at_unix: i64,
+
+    /// Maximum capture duration in milliseconds
+    #[arg(long = "duration-millis")]
+    pub duration_millis: u64,
+
+    /// Sampling interval in microseconds
+    #[arg(long = "sample-period-micros")]
+    pub sample_period_micros: u64,
+
+    /// Confirm this explicit local L3 profile capture
+    #[arg(long = "acknowledge-l3", required = true, action = clap::ArgAction::SetTrue)]
+    pub acknowledge_l3: bool,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+pub enum ConnectProfileTool {
+    Cpu,
+    Memory,
+    Threads,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+pub enum ConnectThreadProfileScope {
+    TokioRuntime,
+    NativeThreads,
 }

 /// `connect register` options
@@ -515,6 +606,8 @@ pub enum CommandResult {
    ConnectRegister(ConnectRegisterOpts),
    /// Local Connect service-license command
    ConnectLicense(ConnectLicenseCommands),
+    /// Consent-bound local Connect profile export
+    ConnectProfile(ConnectProfileOpts),
 }

 /// Create default ServerOpts from environment variables
@@ -651,7 +744,9 @@ mod tests {
        let Some(Commands::Connect(connect)) = cli.command else {
            panic!("connect command expected");
        };
-        let ConnectCommands::Register(register) = connect.command;
+        let ConnectCommands::Register(register) = connect.command else {
+            panic!("connect register command expected");
+        };
        assert_eq!(register.endpoint, "https://connect.example/agent/");
        assert_eq!(register.ca_file, std::path::Path::new("/etc/rustfs/connect-ca.pem"));
        assert_eq!(register.state_dir, std::path::Path::new("/var/lib/rustfs/connect"));
@@ -692,6 +787,48 @@ mod tests {
        assert!(help.to_string().contains("Unix only"));
    }

+    #[test]
+    fn connect_profile_requires_explicit_l3_acknowledgement() {
+        let arguments = [
+            "rustfs",
+            "connect",
+            "profile",
+            "--state-dir",
+            "/var/lib/rustfs/connect",
+            "--output",
+            "/tmp/profile.zip",
+            "--tool",
+            "memory",
+            "--capability",
+            "profile.memory@1",
+            "--organization",
+            "organizations/019e3ae0-0000-7000-8000-000000000001",
+            "--cluster",
+            "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002",
+            "--device",
+            "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002/clusterDevices/019e3ae0-0000-7000-8000-000000000003",
+            "--run-uid",
+            "019e3ae0-0000-7000-8000-000000000004",
+            "--artifact-uid",
+            "019e3ae0-0000-7000-8000-000000000005",
+            "--consent-uid",
+            "019e3ae0-0000-7000-8000-000000000006",
+            "--policy-revision",
+            "1",
+            "--consent-expires-at",
+            "4102444800",
+            "--expires-at",
+            "4102444700",
+            "--duration-millis",
+            "10",
+            "--sample-period-micros",
+            "1000",
+        ];
+        let error = Cli::try_parse_from(arguments).expect_err("an incomplete unacknowledged profile must fail");
+        assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument);
+        assert!(error.to_string().contains("--acknowledge-l3"));
+    }
+
    #[test]
    fn server_help_lists_allocator_reclaim_environment() {
        let result = Cli::try_parse_from(["rustfs", "server", "--help"]);
@@ -52,6 +52,7 @@ mod config_test;
 // Re-export public types
 pub use cli::{CommandResult, InfoOpts, InfoType};
 pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts};
+pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope};
 pub use cli::{DiagnoseFormat, DiagnoseOpts};
 pub use cli::{InspectBucketMetaOpts, InspectCommands, InspectOpts};
 pub use cli::{TlsCommands, TlsInspectOpts, TlsOpts};
@@ -140,6 +140,7 @@ impl Opt {
            Some(Commands::Connect(opts)) => match opts.command {
                ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)),
                ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)),
+                ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)),
            },
            Some(Commands::Server(opts)) => Self::server_command_result(Self::from_server_opts(*opts)),
            None => {
@@ -12,8 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+mod profile_cpu;
+mod profile_memory;
+mod profile_threads;
 mod schedule;

+pub use profile_cpu::{
+    CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION,
+    ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult,
+    ProfileTool, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile,
+    encode_signed_profile_export, export_cpu_profile, save_signed_profile_export,
+};
+pub use profile_memory::export_memory_profile;
+pub use profile_threads::{capture_thread_profile, export_thread_profile};
 pub use schedule::{
    DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus,
    ReceiptOutcome, run_local_environment_once, spawn_environment_schedule,
@@ -0,0 +1,820 @@
+// Copyright 2024 RustFS Team
+//
+// 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.
+
+//! Bounded profile result and signed local export primitives.
+//!
+//! RustFS currently has no reviewed local CPU symbol catalogue or in-process
+//! sampler. [`capture_cpu_profile`] therefore returns an explicit, typed
+//! unsupported result. It never substitutes Pyroscope delivery state, raw
+//! symbol strings, or zero samples for a local CPU measurement.
+
+use std::fs::{self, File, OpenOptions};
+use std::io::{Cursor, Write as _};
+use std::path::Path;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use base64_simd::URL_SAFE_NO_PAD;
+use p256::ecdsa::{Signature, SigningKey, signature::Signer as _};
+use p256::pkcs8::DecodePrivateKey as _;
+use serde::Serialize;
+use sha2::{Digest as _, Sha256};
+use thiserror::Error;
+use time::{OffsetDateTime, format_description::well_known::Rfc3339};
+use tokio_util::sync::CancellationToken;
+use uuid::{Uuid, Variant, Version};
+use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions};
+
+use crate::connect::DeviceIdentity;
+
+pub const PROFILE_SCHEMA_VERSION: u16 = 1;
+pub const CPU_PROFILE_CAPABILITY: &str = "profile.cpu@1";
+pub const MEMORY_PROFILE_CAPABILITY: &str = "profile.memory@1";
+pub const THREAD_PROFILE_CAPABILITY: &str = "profile.threads@1";
+pub const MAX_PROFILE_DURATION: Duration = Duration::from_secs(30);
+pub const MAX_RESULT_BYTES: usize = 262_144;
+pub const MAX_ENVELOPE_BYTES: usize = 16_384;
+pub const MAX_ARCHIVE_BYTES: usize = 524_288;
+pub const MAX_DECOMPRESSED_BYTES: usize = 278_528;
+pub const MAX_BUILD_FEATURES: usize = 64;
+pub const MAX_VALIDITY_SECONDS: i64 = 2_592_000;
+pub const MAX_FUTURE_SKEW_SECONDS: i64 = 300;
+
+const SIGNATURE_DOMAIN: &[u8] = b"rustfs-diagnostic-envelope-v1\0";
+const ENVELOPE_PATH: &str = "envelope.json";
+const SIGNATURE_PATH: &str = "envelope.sig";
+const RESULT_PATH: &str = "result.json";
+const OUTPUT_MODE: u32 = 0o600;
+
+static PROFILE_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false);
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+pub enum ProfileTool {
+    #[serde(rename = "profile.cpu")]
+    Cpu,
+    #[serde(rename = "profile.memory")]
+    Memory,
+    #[serde(rename = "profile.threads")]
+    Threads,
+}
+
+impl ProfileTool {
+    pub const fn id(self) -> &'static str {
+        match self {
+            Self::Cpu => "profile.cpu",
+            Self::Memory => "profile.memory",
+            Self::Threads => "profile.threads",
+        }
+    }
+
+    pub const fn capability(self) -> &'static str {
+        match self {
+            Self::Cpu => CPU_PROFILE_CAPABILITY,
+            Self::Memory => MEMORY_PROFILE_CAPABILITY,
+            Self::Threads => THREAD_PROFILE_CAPABILITY,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub enum ProfileOutcome {
+    Succeeded,
+    Partial,
+    Failed,
+    Unsupported,
+    Cancelled,
+}
+
+impl ProfileOutcome {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Succeeded => "SUCCEEDED",
+            Self::Partial => "PARTIAL",
+            Self::Failed => "FAILED",
+            Self::Unsupported => "UNSUPPORTED",
+            Self::Cancelled => "CANCELLED",
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub enum ProfileReasonCode {
+    Complete,
+    LimitExceeded,
+    SourceUnavailable,
+    UnsupportedTool,
+    UnsupportedVersion,
+    UnsupportedPlatform,
+    Cancelled,
+    CounterReset,
+    InvalidInput,
+    CollectionFailed,
+}
+
+impl ProfileReasonCode {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Complete => "COMPLETE",
+            Self::LimitExceeded => "LIMIT_EXCEEDED",
+            Self::SourceUnavailable => "SOURCE_UNAVAILABLE",
+            Self::UnsupportedTool => "UNSUPPORTED_TOOL",
+            Self::UnsupportedVersion => "UNSUPPORTED_VERSION",
+            Self::UnsupportedPlatform => "UNSUPPORTED_PLATFORM",
+            Self::Cancelled => "CANCELLED",
+            Self::CounterReset => "COUNTER_RESET",
+            Self::InvalidInput => "INVALID_INPUT",
+            Self::CollectionFailed => "COLLECTION_FAILED",
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProfileProvenance {
+    repository: &'static str,
+    source_commit: String,
+    executable_sha256: String,
+    rustfs_version: String,
+    os_family: ProfileOsFamily,
+    architecture: ProfileArchitecture,
+    build_features: Vec<String>,
+}
+
+impl ProfileProvenance {
+    pub fn new(
+        source_commit: impl Into<String>,
+        executable_sha256: impl Into<String>,
+        rustfs_version: impl Into<String>,
+        build_features: Vec<String>,
+    ) -> Self {
+        Self {
+            repository: "rustfs/rustfs",
+            source_commit: source_commit.into(),
+            executable_sha256: executable_sha256.into(),
+            rustfs_version: rustfs_version.into(),
+            os_family: ProfileOsFamily::current(),
+            architecture: ProfileArchitecture::current(),
+            build_features,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub enum ProfileOsFamily {
+    Linux,
+    Darwin,
+    Windows,
+    Freebsd,
+    Other,
+}
+
+impl ProfileOsFamily {
+    fn current() -> Self {
+        match std::env::consts::OS {
+            "linux" => Self::Linux,
+            "macos" => Self::Darwin,
+            "windows" => Self::Windows,
+            "freebsd" => Self::Freebsd,
+            _ => Self::Other,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ProfileArchitecture {
+    #[serde(rename = "x86_64")]
+    X86_64,
+    Aarch64,
+    Other,
+}
+
+impl ProfileArchitecture {
+    fn current() -> Self {
+        match std::env::consts::ARCH {
+            "x86_64" => Self::X86_64,
+            "aarch64" => Self::Aarch64,
+            _ => Self::Other,
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct LocalProfileConsent {
+    pub consent_uid: String,
+    pub policy_revision: u64,
+    pub expires_at_unix: i64,
+    pub confirmed: bool,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ProfileCaptureRequest {
+    pub organization_name: String,
+    pub cluster_name: String,
+    pub device_name: String,
+    pub run_uid: String,
+    pub artifact_uid: String,
+    pub schema_version: u16,
+    pub capability: String,
+    pub consent: LocalProfileConsent,
+    pub produced_at_unix: i64,
+    pub expires_at_unix: i64,
+    pub nonce: [u8; 32],
+    pub duration: Duration,
+    pub sample_period: Duration,
+    pub provenance: ProfileProvenance,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProfileCoverage {
+    requested_units: u32,
+    completed_units: u32,
+    unit: &'static str,
+}
+
+impl ProfileCoverage {
+    pub(super) const fn complete_window() -> Self {
+        Self {
+            requested_units: 1,
+            completed_units: 1,
+            unit: "WINDOW",
+        }
+    }
+
+    const fn none() -> Self {
+        Self {
+            requested_units: 0,
+            completed_units: 0,
+            unit: "WINDOW",
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CpuProfileData {
+    sample_period_micros: u64,
+    samples: Vec<CpuProfileSample>,
+    dropped_sample_count: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CpuProfileSample {
+    symbol_id: String,
+    sample_count: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MemoryProfileData {
+    scope: &'static str,
+    allocated_bytes: u64,
+    allocation_count: u64,
+    sample_period_micros: u64,
+}
+
+impl MemoryProfileData {
+    pub(super) const fn allocation_aggregates(allocated_bytes: u64, allocation_count: u64, sample_period_micros: u64) -> Self {
+        Self {
+            scope: "ALLOCATION_AGGREGATES",
+            allocated_bytes,
+            allocation_count,
+            sample_period_micros,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub enum ThreadProfileScope {
+    TokioRuntime,
+    NativeThreads,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(untagged)]
+pub enum ProfileData {
+    Cpu(CpuProfileData),
+    Memory(MemoryProfileData),
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProfileResult {
+    schema_version: u16,
+    run_uid: String,
+    tool_id: ProfileTool,
+    capability: &'static str,
+    outcome: ProfileOutcome,
+    reason_code: ProfileReasonCode,
+    duration_millis: u64,
+    provenance: ProfileProvenance,
+    coverage: ProfileCoverage,
+    data: Option<ProfileData>,
+}
+
+impl ProfileResult {
+    pub fn outcome(&self) -> ProfileOutcome {
+        self.outcome
+    }
+
+    pub fn reason_code(&self) -> ProfileReasonCode {
+        self.reason_code
+    }
+
+    pub fn data(&self) -> Option<&ProfileData> {
+        self.data.as_ref()
+    }
+
+    pub(super) fn succeeded(request: &ProfileCaptureRequest, tool: ProfileTool, duration: Duration, data: ProfileData) -> Self {
+        Self {
+            schema_version: PROFILE_SCHEMA_VERSION,
+            run_uid: request.run_uid.clone(),
+            tool_id: tool,
+            capability: tool.capability(),
+            outcome: ProfileOutcome::Succeeded,
+            reason_code: ProfileReasonCode::Complete,
+            duration_millis: u64::try_from(duration.as_millis()).unwrap_or(u64::MAX).min(30_000),
+            provenance: request.provenance.clone(),
+            coverage: ProfileCoverage::complete_window(),
+            data: Some(data),
+        }
+    }
+
+    pub(super) fn unsupported(request: &ProfileCaptureRequest, tool: ProfileTool, reason_code: ProfileReasonCode) -> Self {
+        Self {
+            schema_version: PROFILE_SCHEMA_VERSION,
+            run_uid: request.run_uid.clone(),
+            tool_id: tool,
+            capability: tool.capability(),
+            outcome: ProfileOutcome::Unsupported,
+            reason_code,
+            duration_millis: 0,
+            provenance: request.provenance.clone(),
+            coverage: ProfileCoverage::none(),
+            data: None,
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SignedProfileExport {
+    pub artifact_uid: String,
+    pub tool: ProfileTool,
+    pub outcome: ProfileOutcome,
+    pub reason_code: ProfileReasonCode,
+    pub archive_bytes: Vec<u8>,
+    pub archive_sha256: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SavedProfileExport {
+    pub artifact_uid: String,
+    pub archive_size_bytes: u64,
+    pub archive_sha256: String,
+}
+
+#[derive(Debug, Error)]
+pub enum ProfileError {
+    #[error("profile_local_consent_required")]
+    ConsentRequired,
+    #[error("profile_local_consent_expired")]
+    ConsentExpired,
+    #[error("profile_request_expired")]
+    Expired,
+    #[error("profile_invalid_request")]
+    InvalidRequest,
+    #[error("profile_unsupported_version")]
+    UnsupportedVersion,
+    #[error("profile_unsupported_capability")]
+    UnsupportedCapability,
+    #[error("profile_limit_exceeded")]
+    LimitExceeded,
+    #[error("profile_collection_cancelled")]
+    Cancelled,
+    #[error("profile_collection_timed_out")]
+    TimedOut,
+    #[error("profile_collection_already_running")]
+    Busy,
+    #[error("profile_source_unavailable")]
+    SourceUnavailable,
+    #[error("profile_counter_reset")]
+    CounterReset,
+    #[error("profile_export_signing_failed")]
+    Signing,
+    #[error("profile_export_exists")]
+    AlreadyExists,
+    #[error("profile_export_io_failed")]
+    Io(#[source] std::io::Error),
+    #[error("profile_export_encoding_failed")]
+    Encoding,
+    #[error("profile_export_durability_failed_after_commit")]
+    DurabilityAfterCommit(#[source] std::io::Error),
+}
+
+pub fn capture_cpu_profile(request: &ProfileCaptureRequest, cancel: &CancellationToken) -> Result<ProfileResult, ProfileError> {
+    request.validate(ProfileTool::Cpu, unix_now()?)?;
+    check_cancel(cancel)?;
+
+    // The only existing CPU profiler exports to Pyroscope and does not provide
+    // a reviewed local symbol-id catalogue. Publishing samples here would turn
+    // unreviewed process symbols into L3 evidence or invent their mapping.
+    Ok(ProfileResult::unsupported(request, ProfileTool::Cpu, ProfileReasonCode::UnsupportedTool))
+}
+
+pub fn export_cpu_profile(
+    request: &ProfileCaptureRequest,
+    key: &DeviceIdentity,
+    cancel: &CancellationToken,
+) -> Result<SignedProfileExport, ProfileError> {
+    let result = capture_cpu_profile(request, cancel)?;
+    encode_signed_profile_export(request, &result, key, cancel)
+}
+
+pub fn encode_signed_profile_export(
+    request: &ProfileCaptureRequest,
+    result: &ProfileResult,
+    key: &DeviceIdentity,
+    cancel: &CancellationToken,
+) -> Result<SignedProfileExport, ProfileError> {
+    let now = unix_now()?;
+    request.validate(result.tool_id, now)?;
+    check_cancel(cancel)?;
+    let valid_data = match result.outcome {
+        ProfileOutcome::Succeeded | ProfileOutcome::Partial => result.data.is_some(),
+        ProfileOutcome::Failed | ProfileOutcome::Unsupported | ProfileOutcome::Cancelled => result.data.is_none(),
+    };
+    if !valid_data
+        || result.run_uid != request.run_uid
+        || result.schema_version != request.schema_version
+        || result.capability != request.capability
+    {
+        return Err(ProfileError::InvalidRequest);
+    }
+
+    let result_bytes = serde_json::to_vec(result).map_err(|_| ProfileError::Encoding)?;
+    if result_bytes.is_empty() || result_bytes.len() > MAX_RESULT_BYTES {
+        return Err(ProfileError::LimitExceeded);
+    }
+
+    let device_key_id = hex_lower(&Sha256::digest(key.public_key_der()));
+    let envelope = ProfileEnvelope {
+        format_version: "rustfs.connect.diagnosticEnvelope/1",
+        protocol_version: "v1",
+        organization_name: &request.organization_name,
+        cluster_name: &request.cluster_name,
+        device_name: &request.device_name,
+        run_uid: &request.run_uid,
+        artifact_uid: &request.artifact_uid,
+        tool_id: result.tool_id,
+        schema_version: PROFILE_SCHEMA_VERSION,
+        classification: "L3",
+        consent_uid: &request.consent.consent_uid,
+        policy_revision: request.consent.policy_revision,
+        produced_at: timestamp(request.produced_at_unix)?,
+        expires_at: timestamp(request.expires_at_unix)?,
+        nonce: URL_SAFE_NO_PAD.encode_to_string(request.nonce),
+        device_key_id: &device_key_id,
+        payload: ProfilePayload {
+            path: RESULT_PATH,
+            media_type: "application/json",
+            size_bytes: result_bytes.len() as u64,
+            sha256: hex_lower(&Sha256::digest(&result_bytes)),
+        },
+    };
+    let envelope_bytes = serde_json::to_vec(&envelope).map_err(|_| ProfileError::Encoding)?;
+    if envelope_bytes.is_empty() || envelope_bytes.len() > MAX_ENVELOPE_BYTES {
+        return Err(ProfileError::LimitExceeded);
+    }
+    let signature_bytes = signature_document(key, &device_key_id, &envelope_bytes)?;
+    let decompressed = result_bytes
+        .len()
+        .checked_add(envelope_bytes.len())
+        .and_then(|size| size.checked_add(signature_bytes.len()))
+        .ok_or(ProfileError::LimitExceeded)?;
+    if decompressed > MAX_DECOMPRESSED_BYTES {
+        return Err(ProfileError::LimitExceeded);
+    }
+    check_cancel(cancel)?;
+    if unix_now()? >= request.expires_at_unix {
+        return Err(ProfileError::Expired);
+    }
+
+    let archive_bytes = archive(&envelope_bytes, &signature_bytes, &result_bytes)?;
+    if archive_bytes.len() > MAX_ARCHIVE_BYTES {
+        return Err(ProfileError::LimitExceeded);
+    }
+    let archive_sha256 = hex_lower(&Sha256::digest(&archive_bytes));
+
+    Ok(SignedProfileExport {
+        artifact_uid: request.artifact_uid.clone(),
+        tool: result.tool_id,
+        outcome: result.outcome,
+        reason_code: result.reason_code,
+        archive_bytes,
+        archive_sha256,
+    })
+}
+
+pub fn save_signed_profile_export(
+    output: &Path,
+    export: &SignedProfileExport,
+    cancel: &CancellationToken,
+) -> Result<SavedProfileExport, ProfileError> {
+    check_cancel(cancel)?;
+    let parent = output
+        .parent()
+        .filter(|path| !path.as_os_str().is_empty())
+        .unwrap_or_else(|| Path::new("."));
+    let filename = output.file_name().ok_or(ProfileError::InvalidRequest)?.to_string_lossy();
+    let temporary = parent.join(format!(".{filename}.{}.partial", export.artifact_uid));
+    let mut options = OpenOptions::new();
+    options.write(true).create_new(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt as _;
+        options.mode(OUTPUT_MODE);
+    }
+
+    let mut file = options.open(&temporary).map_err(map_create_error)?;
+    let result = (|| {
+        file.write_all(&export.archive_bytes).map_err(ProfileError::Io)?;
+        check_cancel(cancel)?;
+        file.sync_all().map_err(ProfileError::Io)?;
+        check_cancel(cancel)?;
+        fs::hard_link(&temporary, output).map_err(map_publish_error)?;
+        if let Err(error) = fs::remove_file(&temporary) {
+            return Err(ProfileError::DurabilityAfterCommit(error));
+        }
+        #[cfg(unix)]
+        if let Err(error) = File::open(parent).and_then(|directory| directory.sync_all()) {
+            return Err(ProfileError::DurabilityAfterCommit(error));
+        }
+        Ok(SavedProfileExport {
+            artifact_uid: export.artifact_uid.clone(),
+            archive_size_bytes: export.archive_bytes.len() as u64,
+            archive_sha256: export.archive_sha256.clone(),
+        })
+    })();
+    if result.is_err() {
+        let _ = fs::remove_file(&temporary);
+    }
+    result
+}
+
+impl ProfileCaptureRequest {
+    pub(super) fn validate(&self, tool: ProfileTool, now_unix: i64) -> Result<(), ProfileError> {
+        if self.schema_version != PROFILE_SCHEMA_VERSION {
+            return Err(ProfileError::UnsupportedVersion);
+        }
+        if self.capability != tool.capability() {
+            return Err(ProfileError::UnsupportedCapability);
+        }
+        if !self.consent.confirmed || self.consent.policy_revision == 0 {
+            return Err(ProfileError::ConsentRequired);
+        }
+        if self.consent.expires_at_unix <= now_unix || self.expires_at_unix > self.consent.expires_at_unix {
+            return Err(ProfileError::ConsentExpired);
+        }
+        let validity = self
+            .expires_at_unix
+            .checked_sub(self.produced_at_unix)
+            .ok_or(ProfileError::Expired)?;
+        if self.produced_at_unix > now_unix.saturating_add(MAX_FUTURE_SKEW_SECONDS)
+            || validity <= 0
+            || self.expires_at_unix <= now_unix
+            || validity > MAX_VALIDITY_SECONDS
+        {
+            return Err(ProfileError::Expired);
+        }
+        if self.duration.is_zero()
+            || self.duration > MAX_PROFILE_DURATION
+            || self.sample_period.is_zero()
+            || self.sample_period > self.duration
+        {
+            return Err(ProfileError::LimitExceeded);
+        }
+        if !uuid7(&self.run_uid)
+            || !uuid7(&self.artifact_uid)
+            || !uuid7(&self.consent.consent_uid)
+            || !resource_names_match(self)
+            || !lower_hex(&self.provenance.source_commit, 40)
+            || !lower_hex(&self.provenance.executable_sha256, 64)
+            || !version(&self.provenance.rustfs_version)
+            || self.provenance.build_features.len() > MAX_BUILD_FEATURES
+            || !self.provenance.build_features.iter().all(|feature| build_feature(feature))
+        {
+            return Err(ProfileError::InvalidRequest);
+        }
+        Ok(())
+    }
+}
+
+pub(super) struct CollectorLease;
+
+impl CollectorLease {
+    pub(super) fn acquire() -> Result<Self, ProfileError> {
+        PROFILE_COLLECTOR_ACTIVE
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .map(|_| Self)
+            .map_err(|_| ProfileError::Busy)
+    }
+}
+
+impl Drop for CollectorLease {
+    fn drop(&mut self) {
+        PROFILE_COLLECTOR_ACTIVE.store(false, Ordering::Release);
+    }
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ProfileEnvelope<'a> {
+    format_version: &'static str,
+    protocol_version: &'static str,
+    organization_name: &'a str,
+    cluster_name: &'a str,
+    device_name: &'a str,
+    run_uid: &'a str,
+    artifact_uid: &'a str,
+    tool_id: ProfileTool,
+    schema_version: u16,
+    classification: &'static str,
+    consent_uid: &'a str,
+    policy_revision: u64,
+    produced_at: String,
+    expires_at: String,
+    nonce: String,
+    device_key_id: &'a str,
+    payload: ProfilePayload,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ProfilePayload {
+    path: &'static str,
+    media_type: &'static str,
+    size_bytes: u64,
+    sha256: String,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ProfileSignature<'a> {
+    algorithm: &'static str,
+    key_id: &'a str,
+    value: String,
+}
+
+fn signature_document(key: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result<Vec<u8>, ProfileError> {
+    let pkcs8 = key.to_pkcs8_der().map_err(|_| ProfileError::Signing)?;
+    let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| ProfileError::Signing)?;
+    let mut input = Vec::with_capacity(SIGNATURE_DOMAIN.len() + envelope.len());
+    input.extend_from_slice(SIGNATURE_DOMAIN);
+    input.extend_from_slice(envelope);
+    let signature: Signature = signing_key.sign(&input);
+    let value = URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes());
+    serde_json::to_vec(&ProfileSignature {
+        algorithm: "ES256",
+        key_id,
+        value,
+    })
+    .map_err(|_| ProfileError::Encoding)
+}
+
+fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result<Vec<u8>, ProfileError> {
+    let cursor = Cursor::new(Vec::with_capacity(envelope.len() + signature.len() + result.len() + 512));
+    let mut writer = ZipWriter::new(cursor);
+    let options = SimpleFileOptions::DEFAULT
+        .compression_method(CompressionMethod::Stored)
+        .unix_permissions(OUTPUT_MODE);
+    for (name, bytes) in [(ENVELOPE_PATH, envelope), (SIGNATURE_PATH, signature), (RESULT_PATH, result)] {
+        writer.start_file(name, options).map_err(|_| ProfileError::Encoding)?;
+        writer.write_all(bytes).map_err(ProfileError::Io)?;
+    }
+    writer
+        .finish()
+        .map(|cursor| cursor.into_inner())
+        .map_err(|_| ProfileError::Encoding)
+}
+
+fn resource_names_match(request: &ProfileCaptureRequest) -> bool {
+    let Some(organization_uid) = request.organization_name.strip_prefix("organizations/") else {
+        return false;
+    };
+    if !uuid7(organization_uid) {
+        return false;
+    }
+    let cluster_prefix = format!("{}/clusters/", request.organization_name);
+    let Some(cluster_uid) = request.cluster_name.strip_prefix(&cluster_prefix) else {
+        return false;
+    };
+    if !uuid7(cluster_uid) {
+        return false;
+    }
+    let device_prefix = format!("{}/clusterDevices/", request.cluster_name);
+    request.device_name.strip_prefix(&device_prefix).is_some_and(uuid7)
+}
+
+fn uuid7(value: &str) -> bool {
+    Uuid::parse_str(value).is_ok_and(|uuid| {
+        uuid.get_version() == Some(Version::SortRand) && uuid.get_variant() == Variant::RFC4122 && uuid.to_string() == value
+    })
+}
+
+fn lower_hex(value: &str, length: usize) -> bool {
+    value.len() == length
+        && value
+            .bytes()
+            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+}
+
+fn build_feature(value: &str) -> bool {
+    value.len() <= 64
+        && value.as_bytes().first().is_some_and(u8::is_ascii_lowercase)
+        && value
+            .bytes()
+            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-'))
+}
+
+fn version(value: &str) -> bool {
+    if value.is_empty()
+        || value.len() > 64
+        || !value
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-'))
+    {
+        return false;
+    }
+    let (core, suffix) = value
+        .split_once('-')
+        .map_or((value, None), |(core, suffix)| (core, Some(suffix)));
+    if suffix.is_some_and(str::is_empty) {
+        return false;
+    }
+    let mut parts = core.split('.');
+    parts.clone().count() == 3 && parts.all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit()))
+}
+
+fn timestamp(unix: i64) -> Result<String, ProfileError> {
+    OffsetDateTime::from_unix_timestamp(unix)
+        .map_err(|_| ProfileError::InvalidRequest)?
+        .format(&Rfc3339)
+        .map_err(|_| ProfileError::InvalidRequest)
+}
+
+pub(super) fn unix_now() -> Result<i64, ProfileError> {
+    let duration = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(|_| ProfileError::InvalidRequest)?;
+    i64::try_from(duration.as_secs()).map_err(|_| ProfileError::InvalidRequest)
+}
+
+pub(super) fn check_cancel(cancel: &CancellationToken) -> Result<(), ProfileError> {
+    if cancel.is_cancelled() {
+        Err(ProfileError::Cancelled)
+    } else {
+        Ok(())
+    }
+}
+
+fn hex_lower(bytes: &[u8]) -> String {
+    let mut value = String::with_capacity(bytes.len() * 2);
+    for byte in bytes {
+        use std::fmt::Write as _;
+        write!(&mut value, "{byte:02x}").expect("writing hexadecimal to a string cannot fail");
+    }
+    value
+}
+
+fn map_create_error(error: std::io::Error) -> ProfileError {
+    if error.kind() == std::io::ErrorKind::AlreadyExists {
+        ProfileError::AlreadyExists
+    } else {
+        ProfileError::Io(error)
+    }
+}
+
+fn map_publish_error(error: std::io::Error) -> ProfileError {
+    if error.kind() == std::io::ErrorKind::AlreadyExists {
+        ProfileError::AlreadyExists
+    } else {
+        ProfileError::Io(error)
+    }
+}
@@ -0,0 +1,160 @@
+// Copyright 2024 RustFS Team
+//
+// 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.
+
+//! Mimalloc allocation aggregates for the bounded profile contract.
+//!
+//! Only cumulative allocated octets and allocation counts are sampled. Heap
+//! bytes, addresses, stack traces, paths, symbols, and allocator debug text are
+//! excluded from the result.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::time::{Duration, Instant};
+
+use serde_json::Value;
+use tokio_util::sync::CancellationToken;
+
+use crate::connect::DeviceIdentity;
+
+use super::profile_cpu::{
+    CollectorLease, MemoryProfileData, ProfileCaptureRequest, ProfileData, ProfileError, ProfileResult, ProfileTool,
+    SignedProfileExport, check_cancel, encode_signed_profile_export, unix_now,
+};
+
+const MAX_ALLOCATOR_STATS_BYTES: usize = 262_144;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct AllocationSnapshot {
+    total_allocated_bytes: u64,
+    allocation_count: u64,
+}
+
+pub(crate) trait AllocationProfileSource: Send + Sync {
+    fn snapshot(&self) -> Result<AllocationSnapshot, ProfileError>;
+}
+
+struct MimallocProfileSource;
+
+impl AllocationProfileSource for MimallocProfileSource {
+    fn snapshot(&self) -> Result<AllocationSnapshot, ProfileError> {
+        #[cfg(target_os = "windows")]
+        {
+            Err(ProfileError::SourceUnavailable)
+        }
+        #[cfg(not(target_os = "windows"))]
+        {
+            parse_allocator_stats(&rustfs_mimalloc::MiMalloc::stats_json())
+        }
+    }
+}
+
+pub async fn export_memory_profile(
+    request: &ProfileCaptureRequest,
+    key: &DeviceIdentity,
+    cancel: &CancellationToken,
+) -> Result<SignedProfileExport, ProfileError> {
+    export_memory_profile_from(request, key, cancel, &MimallocProfileSource).await
+}
+
+pub(crate) async fn export_memory_profile_from(
+    request: &ProfileCaptureRequest,
+    key: &DeviceIdentity,
+    cancel: &CancellationToken,
+    source: &dyn AllocationProfileSource,
+) -> Result<SignedProfileExport, ProfileError> {
+    request.validate(ProfileTool::Memory, unix_now()?)?;
+    check_cancel(cancel)?;
+    let _lease = CollectorLease::acquire()?;
+    let started = Instant::now();
+    let result = collect_window(request.sample_period, cancel, source);
+    let (before, after) = tokio::time::timeout(request.duration, result)
+        .await
+        .map_err(|_| ProfileError::TimedOut)??;
+    check_cancel(cancel)?;
+
+    let allocated_bytes = after
+        .total_allocated_bytes
+        .checked_sub(before.total_allocated_bytes)
+        .ok_or(ProfileError::CounterReset)?;
+    let allocation_count = after
+        .allocation_count
+        .checked_sub(before.allocation_count)
+        .ok_or(ProfileError::CounterReset)?;
+    let elapsed = started.elapsed();
+    let sample_period_micros = u64::try_from(elapsed.as_micros()).map_err(|_| ProfileError::LimitExceeded)?;
+    let data = MemoryProfileData::allocation_aggregates(allocated_bytes, allocation_count, sample_period_micros.max(1));
+    let result = ProfileResult::succeeded(request, ProfileTool::Memory, elapsed, ProfileData::Memory(data));
+
+    encode_signed_profile_export(request, &result, key, cancel)
+}
+
+fn collect_window<'a>(
+    sample_period: Duration,
+    cancel: &'a CancellationToken,
+    source: &'a dyn AllocationProfileSource,
+) -> Pin<Box<dyn Future<Output = Result<(AllocationSnapshot, AllocationSnapshot), ProfileError>> + Send + 'a>> {
+    Box::pin(async move {
+        let before = source.snapshot()?;
+        tokio::select! {
+            () = cancel.cancelled() => return Err(ProfileError::Cancelled),
+            () = tokio::time::sleep(sample_period) => {}
+        }
+        let after = source.snapshot()?;
+        Ok((before, after))
+    })
+}
+
+pub(crate) fn parse_allocator_stats(stats: &str) -> Result<AllocationSnapshot, ProfileError> {
+    if stats.is_empty() || stats.len() > MAX_ALLOCATOR_STATS_BYTES {
+        return Err(ProfileError::SourceUnavailable);
+    }
+    let value = serde_json::from_str::<Value>(stats).map_err(|_| ProfileError::SourceUnavailable)?;
+    let total_allocated_bytes = sum_metrics(&value, &["malloc_normal", "malloc_huge"], "total")?;
+    let allocation_count = sum_metrics(&value, &["malloc_normal_count", "malloc_huge_count"], "total")?;
+    Ok(AllocationSnapshot {
+        total_allocated_bytes,
+        allocation_count,
+    })
+}
+
+fn sum_metrics(value: &Value, metrics: &[&str], field: &str) -> Result<u64, ProfileError> {
+    metrics.iter().try_fold(0_u64, |sum, metric| {
+        let value = metric_field(value, metric, field).ok_or(ProfileError::SourceUnavailable)?;
+        sum.checked_add(value).ok_or(ProfileError::LimitExceeded)
+    })
+}
+
+fn metric_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
+    match value {
+        Value::Object(fields) => {
+            if let Some(metric_value) = fields.get(metric)
+                && let Some(value) = numeric_field(metric_value, field)
+            {
+                return Some(value);
+            }
+            fields.values().find_map(|value| metric_field(value, metric, field))
+        }
+        Value::Array(values) => values.iter().find_map(|value| metric_field(value, metric, field)),
+        _ => None,
+    }
+}
+
+fn numeric_field(value: &Value, field: &str) -> Option<u64> {
+    match value {
+        Value::Number(number) => number.as_u64(),
+        Value::String(value) => value.parse().ok(),
+        Value::Object(fields) => fields.get(field).and_then(|value| numeric_field(value, field)),
+        _ => None,
+    }
+}
@@ -0,0 +1,54 @@
+// Copyright 2024 RustFS Team
+//
+// 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.
+
+//! Explicit thread/runtime profile capability result.
+//!
+//! Dial9 currently exposes session and disk-buffer state, not bounded counts
+//! for RUNNABLE, WAITING, BLOCKED, and UNKNOWN threads. Native thread state
+//! would require a separate reviewed platform adapter. Neither source is
+//! relabelled as the contract's thread data.
+
+use tokio_util::sync::CancellationToken;
+
+use super::profile_cpu::{
+    ProfileCaptureRequest, ProfileError, ProfileReasonCode, ProfileResult, ProfileTool, ThreadProfileScope, check_cancel,
+    encode_signed_profile_export, unix_now,
+};
+use crate::connect::DeviceIdentity;
+
+use super::profile_cpu::SignedProfileExport;
+
+pub fn capture_thread_profile(
+    request: &ProfileCaptureRequest,
+    _scope: ThreadProfileScope,
+    cancel: &CancellationToken,
+) -> Result<ProfileResult, ProfileError> {
+    request.validate(ProfileTool::Threads, unix_now()?)?;
+    check_cancel(cancel)?;
+    Ok(ProfileResult::unsupported(
+        request,
+        ProfileTool::Threads,
+        ProfileReasonCode::UnsupportedTool,
+    ))
+}
+
+pub fn export_thread_profile(
+    request: &ProfileCaptureRequest,
+    scope: ThreadProfileScope,
+    key: &DeviceIdentity,
+    cancel: &CancellationToken,
+) -> Result<SignedProfileExport, ProfileError> {
+    let result = capture_thread_profile(request, scope, cancel)?;
+    encode_signed_profile_export(request, &result, key, cancel)
+}
@@ -45,8 +45,12 @@ pub use client::{ClientError, ConnectClient, ConnectConfig};
 pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule};
 pub use credential_store::{CredentialStore, DeviceCredential};
 pub use diagnostics::{
-    DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus,
-    ReceiptOutcome, run_local_environment_once, spawn_environment_schedule,
+    CPU_PROFILE_CAPABILITY, DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime,
+    DiagnosticScheduleStatus, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION,
+    ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult,
+    ProfileTool, ReceiptOutcome, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope,
+    capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, export_cpu_profile, export_memory_profile,
+    export_thread_profile, run_local_environment_once, save_signed_profile_export, spawn_environment_schedule,
 };
 pub use environment::{
    ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError,
@@ -13,7 +13,10 @@
 // limitations under the License.

 use crate::{
-    config::{CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, Opt},
+    config::{
+        CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectProfileOpts, ConnectProfileTool,
+        ConnectThreadProfileScope, Opt,
+    },
    startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle},
    startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight},
    startup_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context},
@@ -22,7 +25,8 @@ use crate::{
    storage_api::server::http::ServerContextSlot,
    storage_api::startup::storage::bootstrap_instance_ctx,
 };
-use std::io::{Error, Result};
+use std::io::{Error, Read as _, Result};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tracing::{error, instrument};

 const LOG_COMPONENT_MAIN: &str = "main";
@@ -130,6 +134,7 @@ async fn async_main() -> Result<()> {
            return Ok(());
        }
        CommandResult::ConnectLicense(command) => return execute_connect_license(command),
+        CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await,
        CommandResult::Server(config) => config,
    };

@@ -159,6 +164,173 @@ async fn async_main() -> Result<()> {
    }
 }

+async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> {
+    use crate::connect::{
+        IdentityStore, LocalProfileConsent, ProfileCaptureRequest, ProfileProvenance, ThreadProfileScope, export_cpu_profile,
+        export_memory_profile, export_thread_profile, save_signed_profile_export,
+    };
+    use rand::{TryRng as _, rngs::SysRng};
+
+    let key = IdentityStore::new(options.state_dir.join("identity"))
+        .load()
+        .map_err(Error::other)?
+        .ok_or_else(|| Error::other("connect profile requires an enrolled device identity"))?;
+    let executable_sha256 = hash_current_executable()?;
+    let produced_at_unix = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(Error::other)
+        .and_then(|duration| i64::try_from(duration.as_secs()).map_err(Error::other))?;
+    let mut nonce = [0_u8; 32];
+    SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?;
+    let request = ProfileCaptureRequest {
+        organization_name: options.organization,
+        cluster_name: options.cluster,
+        device_name: options.device,
+        run_uid: options.run_uid,
+        artifact_uid: options.artifact_uid,
+        schema_version: options.schema_version,
+        capability: options.capability,
+        consent: LocalProfileConsent {
+            consent_uid: options.consent_uid,
+            policy_revision: options.policy_revision,
+            expires_at_unix: options.consent_expires_at_unix,
+            confirmed: options.acknowledge_l3,
+        },
+        produced_at_unix,
+        expires_at_unix: options.expires_at_unix,
+        nonce,
+        duration: Duration::from_millis(options.duration_millis),
+        sample_period: Duration::from_micros(options.sample_period_micros),
+        provenance: ProfileProvenance::new(
+            crate::version::build::COMMIT_HASH,
+            executable_sha256,
+            env!("CARGO_PKG_VERSION"),
+            enabled_build_features(),
+        ),
+    };
+    let cancel = tokio_util::sync::CancellationToken::new();
+    let capture = async {
+        match options.tool {
+            ConnectProfileTool::Cpu => {
+                if options.thread_scope.is_some() {
+                    return Err(Error::other("--thread-scope is valid only for the threads profile"));
+                }
+                export_cpu_profile(&request, &key, &cancel).map_err(Error::other)
+            }
+            ConnectProfileTool::Memory => {
+                if options.thread_scope.is_some() {
+                    return Err(Error::other("--thread-scope is valid only for the threads profile"));
+                }
+                export_memory_profile(&request, &key, &cancel).await.map_err(Error::other)
+            }
+            ConnectProfileTool::Threads => {
+                let scope = match options.thread_scope {
+                    Some(ConnectThreadProfileScope::TokioRuntime) => ThreadProfileScope::TokioRuntime,
+                    Some(ConnectThreadProfileScope::NativeThreads) => ThreadProfileScope::NativeThreads,
+                    None => return Err(Error::other("--thread-scope is required for the threads profile")),
+                };
+                export_thread_profile(&request, scope, &key, &cancel).map_err(Error::other)
+            }
+        }
+    };
+    tokio::pin!(capture);
+    let export = tokio::select! {
+        biased;
+        signal = tokio::signal::ctrl_c() => {
+            signal.map_err(Error::other)?;
+            cancel.cancel();
+            return Err(Error::other("profile collection cancelled"));
+        }
+        result = capture.as_mut() => result?,
+    };
+    drop(capture);
+    let tool = export.tool;
+    let outcome = export.outcome;
+    let reason_code = export.reason_code;
+    let output = options.output;
+    let writer_cancel = cancel.clone();
+    let mut writer = tokio::task::spawn_blocking(move || save_signed_profile_export(&output, &export, &writer_cancel));
+    let receipt = tokio::select! {
+        biased;
+        signal = tokio::signal::ctrl_c() => {
+            signal.map_err(Error::other)?;
+            cancel.cancel();
+            writer.await.map_err(Error::other)?.map_err(Error::other)?
+        }
+        result = &mut writer => result.map_err(Error::other)?.map_err(Error::other)?,
+    };
+
+    println!("tool={} outcome={} reason={}", tool.id(), outcome.as_str(), reason_code.as_str());
+    println!(
+        "artifact={} bytes={} sha256={}",
+        receipt.artifact_uid, receipt.archive_size_bytes, receipt.archive_sha256
+    );
+    println!("upload=not-performed");
+    Ok(())
+}
+
+fn hash_current_executable() -> Result<String> {
+    use sha2::{Digest as _, Sha256};
+
+    const MAX_EXECUTABLE_BYTES: u64 = 1_073_741_824;
+    let path = std::env::current_exe().map_err(Error::other)?;
+    let mut file = std::fs::File::open(path).map_err(Error::other)?;
+    let metadata = file.metadata().map_err(Error::other)?;
+    if !metadata.is_file() || metadata.len() == 0 || metadata.len() > MAX_EXECUTABLE_BYTES {
+        return Err(Error::other("current executable is outside the profile provenance limit"));
+    }
+    let mut hasher = Sha256::new();
+    let mut buffer = [0_u8; 64 * 1024];
+    let mut read_bytes = 0_u64;
+    loop {
+        let count = file.read(&mut buffer).map_err(Error::other)?;
+        if count == 0 {
+            break;
+        }
+        read_bytes = read_bytes
+            .checked_add(u64::try_from(count).map_err(Error::other)?)
+            .ok_or_else(|| Error::other("current executable is outside the profile provenance limit"))?;
+        if read_bytes > MAX_EXECUTABLE_BYTES {
+            return Err(Error::other("current executable is outside the profile provenance limit"));
+        }
+        hasher.update(&buffer[..count]);
+    }
+    if read_bytes != metadata.len() {
+        return Err(Error::other("current executable changed while hashing profile provenance"));
+    }
+    Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower))
+}
+
+fn enabled_build_features() -> Vec<String> {
+    let mut features = Vec::new();
+    for (enabled, name) in [
+        (cfg!(feature = "connect-e2e-short-credentials"), "connect-e2e-short-credentials"),
+        (cfg!(feature = "dial9"), "dial9"),
+        (cfg!(feature = "e2e-test-hooks"), "e2e-test-hooks"),
+        (cfg!(feature = "ftps"), "ftps"),
+        (cfg!(feature = "full"), "full"),
+        (cfg!(feature = "gcs"), "gcs"),
+        (cfg!(feature = "hotpath"), "hotpath"),
+        (cfg!(feature = "hotpath-alloc"), "hotpath-alloc"),
+        (cfg!(feature = "hotpath-cpu"), "hotpath-cpu"),
+        (cfg!(feature = "io-scheduler-debug"), "io-scheduler-debug"),
+        (cfg!(feature = "license"), "license"),
+        (cfg!(feature = "metrics-gpu"), "metrics-gpu"),
+        (cfg!(feature = "offline-enrollment-e2e-root"), "offline-enrollment-e2e-root"),
+        (cfg!(feature = "pyroscope"), "pyroscope"),
+        (cfg!(feature = "rio-v2"), "rio-v2"),
+        (cfg!(feature = "sftp"), "sftp"),
+        (cfg!(feature = "swift"), "swift"),
+        (cfg!(feature = "tracing-chunk-debug"), "tracing-chunk-debug"),
+        (cfg!(feature = "webdav"), "webdav"),
+    ] {
+        if enabled {
+            features.push(name.to_owned());
+        }
+    }
+    features
+}
+
 fn execute_connect_license(command: ConnectLicenseCommands) -> Result<()> {
    use crate::connect::{apply_license_artifact, inspect_installed_license, verify_license_artifact};

@@ -0,0 +1,157 @@
+// Copyright 2024 RustFS Team
+//
+// 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.
+
+mod connect {
+    pub use rustfs::connect::DeviceIdentity;
+}
+
+#[allow(dead_code)]
+#[path = "../src/connect/diagnostics/profile_cpu.rs"]
+mod profile_cpu;
+
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use profile_cpu::{
+    CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, ProfileCaptureRequest, ProfileError, ProfileOutcome,
+    ProfileProvenance, ProfileReasonCode, capture_cpu_profile,
+};
+use tokio_util::sync::CancellationToken;
+
+fn now() -> i64 {
+    SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64
+}
+
+fn request() -> ProfileCaptureRequest {
+    let now = now();
+    let organization = "organizations/019e3ae0-0000-7000-8000-000000000001";
+    let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000002");
+    ProfileCaptureRequest {
+        organization_name: organization.to_string(),
+        cluster_name: cluster.clone(),
+        device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000003"),
+        run_uid: "019e3ae0-0000-7000-8000-000000000004".to_string(),
+        artifact_uid: "019e3ae0-0000-7000-8000-000000000005".to_string(),
+        schema_version: 1,
+        capability: CPU_PROFILE_CAPABILITY.to_string(),
+        consent: LocalProfileConsent {
+            consent_uid: "019e3ae0-0000-7000-8000-000000000006".to_string(),
+            policy_revision: 1,
+            expires_at_unix: now + 120,
+            confirmed: true,
+        },
+        produced_at_unix: now,
+        expires_at_unix: now + 60,
+        nonce: [0x5a; 32],
+        duration: Duration::from_secs(1),
+        sample_period: Duration::from_millis(10),
+        provenance: ProfileProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec!["default".to_string()]),
+    }
+}
+
+#[test]
+fn cpu_without_reviewed_symbol_catalog_is_explicitly_unsupported() {
+    let result = capture_cpu_profile(&request(), &CancellationToken::new()).expect("unsupported is a typed result");
+    assert_eq!(result.outcome(), ProfileOutcome::Unsupported);
+    assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool);
+    assert!(result.data().is_none(), "unsupported CPU must not fabricate zero samples");
+
+    let json = serde_json::to_value(result).expect("result JSON");
+    assert_eq!(json["toolId"], "profile.cpu");
+    assert_eq!(json["capability"], "profile.cpu@1");
+    assert_eq!(json["coverage"]["requestedUnits"], 0);
+    assert_eq!(json["coverage"]["completedUnits"], 0);
+    assert!(json["data"].is_null());
+}
+
+#[test]
+fn cpu_refuses_missing_or_expired_local_consent_before_capability_disclosure() {
+    let mut missing = request();
+    missing.consent.confirmed = false;
+    assert!(matches!(
+        capture_cpu_profile(&missing, &CancellationToken::new()),
+        Err(ProfileError::ConsentRequired)
+    ));
+
+    let mut expired = request();
+    expired.consent.expires_at_unix = now() - 1;
+    assert!(matches!(
+        capture_cpu_profile(&expired, &CancellationToken::new()),
+        Err(ProfileError::ConsentExpired)
+    ));
+}
+
+#[test]
+fn cpu_negotiation_and_resource_limits_are_closed_at_the_boundary() {
+    let mut invalid = request();
+    invalid.schema_version = 2;
+    assert!(matches!(
+        capture_cpu_profile(&invalid, &CancellationToken::new()),
+        Err(ProfileError::UnsupportedVersion)
+    ));
+
+    let mut invalid = request();
+    invalid.capability = "profile.cpu@2".to_string();
+    assert!(matches!(
+        capture_cpu_profile(&invalid, &CancellationToken::new()),
+        Err(ProfileError::UnsupportedCapability)
+    ));
+
+    let mut boundary = request();
+    boundary.duration = MAX_PROFILE_DURATION;
+    boundary.sample_period = MAX_PROFILE_DURATION;
+    assert!(capture_cpu_profile(&boundary, &CancellationToken::new()).is_ok());
+
+    let mut over = request();
+    over.duration = MAX_PROFILE_DURATION + Duration::from_nanos(1);
+    assert!(matches!(
+        capture_cpu_profile(&over, &CancellationToken::new()),
+        Err(ProfileError::LimitExceeded)
+    ));
+
+    let mut features = request();
+    features.provenance = ProfileProvenance::new(
+        "a".repeat(40),
+        "b".repeat(64),
+        "1.0.0",
+        (0..65).map(|index| format!("feature_{index}")).collect(),
+    );
+    assert!(matches!(
+        capture_cpu_profile(&features, &CancellationToken::new()),
+        Err(ProfileError::InvalidRequest)
+    ));
+
+    let mut empty_version_suffix = request();
+    empty_version_suffix.provenance =
+        ProfileProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-", vec!["default".to_string()]);
+    assert!(matches!(
+        capture_cpu_profile(&empty_version_suffix, &CancellationToken::new()),
+        Err(ProfileError::InvalidRequest)
+    ));
+
+    let mut overflowing_window = request();
+    overflowing_window.produced_at_unix = i64::MIN;
+    overflowing_window.expires_at_unix = i64::MAX;
+    overflowing_window.consent.expires_at_unix = i64::MAX;
+    assert!(matches!(
+        capture_cpu_profile(&overflowing_window, &CancellationToken::new()),
+        Err(ProfileError::Expired)
+    ));
+}
+
+#[test]
+fn cpu_honors_pre_cancelled_capture() {
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    assert!(matches!(capture_cpu_profile(&request(), &cancel), Err(ProfileError::Cancelled)));
+}
@@ -0,0 +1,270 @@
+// Copyright 2024 RustFS Team
+//
+// 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.
+
+mod connect {
+    pub use rustfs::connect::DeviceIdentity;
+}
+
+#[allow(dead_code)]
+#[path = "../src/connect/diagnostics/profile_cpu.rs"]
+mod profile_cpu;
+#[path = "../src/connect/diagnostics/profile_memory.rs"]
+mod profile_memory;
+
+use std::fs;
+use std::io::{Cursor, Read as _};
+#[cfg(unix)]
+use std::os::unix::fs::PermissionsExt as _;
+use std::sync::Mutex;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use base64_simd::URL_SAFE_NO_PAD;
+use p256::ecdsa::signature::Verifier as _;
+use p256::ecdsa::{Signature, VerifyingKey};
+use p256::pkcs8::DecodePublicKey as _;
+use profile_cpu::{
+    LocalProfileConsent, MEMORY_PROFILE_CAPABILITY, ProfileCaptureRequest, ProfileError, ProfileProvenance,
+    save_signed_profile_export,
+};
+use profile_memory::{AllocationProfileSource, export_memory_profile, export_memory_profile_from, parse_allocator_stats};
+use sha2::{Digest as _, Sha256};
+use tokio_util::sync::CancellationToken;
+use zip::ZipArchive;
+
+static TEST_PROFILE_LOCK: Mutex<()> = Mutex::new(());
+
+fn profile_test_lock() -> std::sync::MutexGuard<'static, ()> {
+    TEST_PROFILE_LOCK.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+#[global_allocator]
+static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
+
+fn now() -> i64 {
+    SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64
+}
+
+fn request() -> ProfileCaptureRequest {
+    let now = now();
+    let organization = "organizations/019e3ae0-0000-7000-8000-000000000011";
+    let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000012");
+    ProfileCaptureRequest {
+        organization_name: organization.to_string(),
+        cluster_name: cluster.clone(),
+        device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000013"),
+        run_uid: "019e3ae0-0000-7000-8000-000000000014".to_string(),
+        artifact_uid: "019e3ae0-0000-7000-8000-000000000015".to_string(),
+        schema_version: 1,
+        capability: MEMORY_PROFILE_CAPABILITY.to_string(),
+        consent: LocalProfileConsent {
+            consent_uid: "019e3ae0-0000-7000-8000-000000000016".to_string(),
+            policy_revision: 7,
+            expires_at_unix: now + 120,
+            confirmed: true,
+        },
+        produced_at_unix: now,
+        expires_at_unix: now + 60,
+        nonce: [0x6b; 32],
+        duration: Duration::from_secs(1),
+        sample_period: Duration::from_millis(10),
+        provenance: ProfileProvenance::new("c".repeat(40), "d".repeat(64), "1.0.0-rc.6", vec!["default".to_string()]),
+    }
+}
+
+struct SequenceSource {
+    values: Mutex<Vec<&'static str>>,
+}
+
+impl SequenceSource {
+    fn new(first: &'static str, second: &'static str) -> Self {
+        Self {
+            values: Mutex::new(vec![second, first]),
+        }
+    }
+}
+
+impl AllocationProfileSource for SequenceSource {
+    fn snapshot(&self) -> Result<profile_memory::AllocationSnapshot, ProfileError> {
+        let value = self.values.lock().expect("sequence lock").pop().expect("two samples");
+        parse_allocator_stats(value)
+    }
+}
+
+fn stats(bytes: u64, count: u64) -> String {
+    format!(
+        r#"{{"malloc_normal":{{"total":{bytes}}},"malloc_huge":{{"total":0}},"malloc_normal_count":{{"total":{count}}},"malloc_huge_count":{{"total":0}}}}"#
+    )
+}
+
+fn archive_entry(archive: &mut ZipArchive<Cursor<Vec<u8>>>, name: &str) -> Vec<u8> {
+    let mut entry = archive.by_name(name).expect("archive entry");
+    let mut bytes = Vec::new();
+    entry.read_to_end(&mut bytes).expect("read archive entry");
+    bytes
+}
+
+#[tokio::test]
+async fn real_mimalloc_profile_produces_a_signed_three_file_export() {
+    let _guard = profile_test_lock();
+    let key = connect::DeviceIdentity::generate();
+    let export = export_memory_profile(&request(), &key, &CancellationToken::new())
+        .await
+        .expect("real allocation aggregate export");
+    assert!(export.archive_bytes.len() <= profile_cpu::MAX_ARCHIVE_BYTES);
+    assert_eq!(export.archive_sha256, hex(&Sha256::digest(&export.archive_bytes)));
+
+    let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes.clone())).expect("profile archive");
+    assert_eq!(archive.len(), 3);
+    let envelope_bytes = archive_entry(&mut archive, "envelope.json");
+    let signature_bytes = archive_entry(&mut archive, "envelope.sig");
+    let result_bytes = archive_entry(&mut archive, "result.json");
+    let envelope: serde_json::Value = serde_json::from_slice(&envelope_bytes).expect("envelope JSON");
+    let signature: serde_json::Value = serde_json::from_slice(&signature_bytes).expect("signature JSON");
+    let result: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON");
+
+    assert_eq!(envelope["classification"], "L3");
+    assert_eq!(envelope["payload"]["path"], "result.json");
+    assert_eq!(envelope["payload"]["sizeBytes"], result_bytes.len());
+    assert_eq!(envelope["payload"]["sha256"], hex(&Sha256::digest(&result_bytes)));
+    assert_eq!(result["toolId"], "profile.memory");
+    assert_eq!(result["outcome"], "SUCCEEDED");
+    assert_eq!(result["data"]["scope"], "ALLOCATION_AGGREGATES");
+    assert!(result["data"]["allocatedBytes"].is_u64());
+    assert!(result["data"]["allocationCount"].is_u64());
+    assert!(result["data"]["samplePeriodMicros"].as_u64().is_some_and(|value| value > 0));
+    let encoded = serde_json::to_string(&result).expect("encoded result");
+    for forbidden in ["/Users/", "/proc/", "AKIA", "secret", "stack", "symbol"] {
+        assert!(!encoded.contains(forbidden), "result leaked forbidden material: {forbidden}");
+    }
+
+    assert_eq!(signature["algorithm"], "ES256");
+    let raw_signature = URL_SAFE_NO_PAD
+        .decode_to_vec(signature["value"].as_str().expect("signature value"))
+        .expect("base64url signature");
+    assert_eq!(raw_signature.len(), 64);
+    let signature_value = Signature::from_slice(&raw_signature).expect("P-256 signature");
+    assert_eq!(signature_value.normalize_s(), signature_value);
+    let public = VerifyingKey::from_public_key_der(&key.public_key_der()).expect("public key");
+    let mut input = b"rustfs-diagnostic-envelope-v1\0".to_vec();
+    input.extend_from_slice(&envelope_bytes);
+    public
+        .verify(&input, &signature_value)
+        .expect("signature over exact envelope bytes");
+}
+
+#[tokio::test]
+async fn memory_profile_reports_counter_reset_and_cancellation_without_an_artifact() {
+    let _guard = profile_test_lock();
+    let first = Box::leak(stats(100, 10).into_boxed_str());
+    let second = Box::leak(stats(90, 11).into_boxed_str());
+    let source = SequenceSource::new(first, second);
+    let key = connect::DeviceIdentity::generate();
+    assert!(matches!(
+        export_memory_profile_from(&request(), &key, &CancellationToken::new(), &source).await,
+        Err(ProfileError::CounterReset)
+    ));
+
+    let first = Box::leak(stats(100, 10).into_boxed_str());
+    let second = Box::leak(stats(120, 12).into_boxed_str());
+    let source = SequenceSource::new(first, second);
+    let mut request = request();
+    request.sample_period = Duration::from_secs(1);
+    request.duration = Duration::from_secs(2);
+    let cancel = CancellationToken::new();
+    let cancellation = async {
+        tokio::time::sleep(Duration::from_millis(5)).await;
+        cancel.cancel();
+    };
+    let (result, ()) = tokio::join!(export_memory_profile_from(&request, &key, &cancel, &source), cancellation);
+    assert!(matches!(result, Err(ProfileError::Cancelled)));
+}
+
+#[tokio::test]
+async fn memory_profile_uses_only_bounded_allocator_aggregates() {
+    let _guard = profile_test_lock();
+    let first = Box::leak(stats(1_000, 20).into_boxed_str());
+    let second = Box::leak(stats(1_250, 24).into_boxed_str());
+    let source = SequenceSource::new(first, second);
+    let key = connect::DeviceIdentity::generate();
+    let export = export_memory_profile_from(&request(), &key, &CancellationToken::new(), &source)
+        .await
+        .expect("aggregate export");
+    let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes)).expect("profile archive");
+    let result: serde_json::Value = serde_json::from_slice(&archive_entry(&mut archive, "result.json")).expect("result JSON");
+    assert_eq!(result["data"]["allocatedBytes"], 250);
+    assert_eq!(result["data"]["allocationCount"], 4);
+
+    let oversized = format!("{}{}", stats(1, 1), " ".repeat(262_145));
+    assert!(matches!(parse_allocator_stats(&oversized), Err(ProfileError::SourceUnavailable)));
+}
+
+#[tokio::test]
+async fn memory_profile_allows_only_one_collector_at_a_time() {
+    let _guard = profile_test_lock();
+    let first = Box::leak(stats(100, 10).into_boxed_str());
+    let second = Box::leak(stats(120, 12).into_boxed_str());
+    let source = SequenceSource::new(first, second);
+    let second_source = SequenceSource::new(first, second);
+    let key = connect::DeviceIdentity::generate();
+    let mut request = request();
+    request.sample_period = Duration::from_millis(30);
+
+    let first_cancel = CancellationToken::new();
+    let second_cancel = CancellationToken::new();
+    let first_capture = export_memory_profile_from(&request, &key, &first_cancel, &source);
+    let second_capture = async {
+        tokio::time::sleep(Duration::from_millis(5)).await;
+        export_memory_profile_from(&request, &key, &second_cancel, &second_source).await
+    };
+    let (first_result, second_result) = tokio::join!(first_capture, second_capture);
+    assert!(first_result.is_ok());
+    assert!(matches!(second_result, Err(ProfileError::Busy)));
+}
+
+#[tokio::test]
+async fn signed_export_is_private_no_clobber_and_cancel_safe() {
+    let _guard = profile_test_lock();
+    let first = Box::leak(stats(100, 10).into_boxed_str());
+    let second = Box::leak(stats(150, 12).into_boxed_str());
+    let source = SequenceSource::new(first, second);
+    let key = connect::DeviceIdentity::generate();
+    let export = export_memory_profile_from(&request(), &key, &CancellationToken::new(), &source)
+        .await
+        .expect("aggregate export");
+    let directory = tempfile::tempdir().expect("temporary output");
+    let output = directory.path().join("profile.zip");
+    let receipt = save_signed_profile_export(&output, &export, &CancellationToken::new()).expect("save export");
+    assert_eq!(receipt.archive_sha256, export.archive_sha256);
+    assert_eq!(receipt.archive_size_bytes, export.archive_bytes.len() as u64);
+    assert_eq!(fs::read(&output).expect("saved archive"), export.archive_bytes);
+    #[cfg(unix)]
+    assert_eq!(fs::metadata(&output).expect("metadata").permissions().mode() & 0o777, 0o600);
+    assert!(matches!(
+        save_signed_profile_export(&output, &export, &CancellationToken::new()),
+        Err(ProfileError::AlreadyExists)
+    ));
+
+    let cancelled_output = directory.path().join("cancelled.zip");
+    let cancelled = CancellationToken::new();
+    cancelled.cancel();
+    assert!(matches!(
+        save_signed_profile_export(&cancelled_output, &export, &cancelled),
+        Err(ProfileError::Cancelled)
+    ));
+    assert!(!cancelled_output.exists());
+}
+
+fn hex(bytes: &[u8]) -> String {
+    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+}
@@ -0,0 +1,90 @@
+// Copyright 2024 RustFS Team
+//
+// 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.
+
+mod connect {
+    pub use rustfs::connect::DeviceIdentity;
+}
+
+#[allow(dead_code)]
+#[path = "../src/connect/diagnostics/profile_cpu.rs"]
+mod profile_cpu;
+#[path = "../src/connect/diagnostics/profile_threads.rs"]
+mod profile_threads;
+
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use profile_cpu::{
+    LocalProfileConsent, ProfileCaptureRequest, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode,
+    THREAD_PROFILE_CAPABILITY, ThreadProfileScope,
+};
+use profile_threads::capture_thread_profile;
+use tokio_util::sync::CancellationToken;
+
+fn request() -> ProfileCaptureRequest {
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64;
+    let organization = "organizations/019e3ae0-0000-7000-8000-000000000021";
+    let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000022");
+    ProfileCaptureRequest {
+        organization_name: organization.to_string(),
+        cluster_name: cluster.clone(),
+        device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000023"),
+        run_uid: "019e3ae0-0000-7000-8000-000000000024".to_string(),
+        artifact_uid: "019e3ae0-0000-7000-8000-000000000025".to_string(),
+        schema_version: 1,
+        capability: THREAD_PROFILE_CAPABILITY.to_string(),
+        consent: LocalProfileConsent {
+            consent_uid: "019e3ae0-0000-7000-8000-000000000026".to_string(),
+            policy_revision: 1,
+            expires_at_unix: now + 120,
+            confirmed: true,
+        },
+        produced_at_unix: now,
+        expires_at_unix: now + 60,
+        nonce: [0x7c; 32],
+        duration: Duration::from_secs(1),
+        sample_period: Duration::from_millis(10),
+        provenance: ProfileProvenance::new("e".repeat(40), "f".repeat(64), "1.0.0-rc.6", vec!["default".to_string()]),
+    }
+}
+
+#[test]
+fn tokio_and_native_thread_scopes_remain_explicitly_unsupported() {
+    for scope in [ThreadProfileScope::TokioRuntime, ThreadProfileScope::NativeThreads] {
+        let result = capture_thread_profile(&request(), scope, &CancellationToken::new()).expect("unsupported result");
+        assert_eq!(result.outcome(), ProfileOutcome::Unsupported);
+        assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool);
+        assert!(result.data().is_none(), "unsupported scope must not publish zero state counts");
+        let json = serde_json::to_value(result).expect("result JSON");
+        assert_eq!(json["toolId"], "profile.threads");
+        assert_eq!(json["capability"], "profile.threads@1");
+        assert!(json["data"].is_null());
+    }
+}
+
+#[test]
+fn thread_profile_rejects_wrong_negotiation_and_cancellation() {
+    let mut invalid = request();
+    invalid.capability = "profile.threads@2".to_string();
+    assert!(matches!(
+        capture_thread_profile(&invalid, ThreadProfileScope::TokioRuntime, &CancellationToken::new()),
+        Err(ProfileError::UnsupportedCapability)
+    ));
+
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    assert!(matches!(
+        capture_thread_profile(&request(), ThreadProfileScope::NativeThreads, &cancel),
+        Err(ProfileError::Cancelled)
+    ));
+}