исправить: закрыть ревью критических ошибок

.gitea/workflows/ci.yml

@@ -237,15 +237,19 @@ jobs:
           EOF
           docker compose -f deploy/community/docker-compose.images.yml \
             --env-file .tmp/community-smoke.env --profile local-db up -d --wait
-          ui_address="$(docker compose -f deploy/community/docker-compose.images.yml \
-            --env-file .tmp/community-smoke.env port ui 3000)"
-          printf 'http://%s\n' "$ui_address" > .tmp/community-smoke.url
 
       - name: Run authenticated Community image smoke
         env:
           CRANK_STAGING_ADMIN_EMAIL: owner@crank.test
           CRANK_STAGING_ADMIN_PASSWORD: ci-admin-password
-        run: scripts/authenticated-product-smoke.sh "$(cat .tmp/community-smoke.url)"
+        run: |
+          set -eu
+          project_name="$(sed -n 's/^COMPOSE_PROJECT_NAME=//p' .tmp/community-smoke.env)"
+          docker run --rm --network "${project_name}_default" \
+            -e CRANK_STAGING_ADMIN_EMAIL \
+            -e CRANK_STAGING_ADMIN_PASSWORD \
+            -i python:3.13-alpine \
+            python - http://ui:3000 < scripts/authenticated-product-smoke.py
 
       - name: Show Community image logs
         if: failure()

Cargo.lock

@@ -37,6 +37,7 @@ dependencies = [
  "crank-test-support",
  "crank-trace",
  "metrics",
+ "metrics-util",
  "opentelemetry",
  "opentelemetry_sdk",
  "rand 0.10.2",
@@ -654,9 +655,9 @@ dependencies = [
  "async-trait",
  "axum",
  "crank-core",
+ "crank-metrics",
  "crank-trace",
  "futures-util",
- "metrics",
  "opentelemetry",
  "opentelemetry_sdk",
  "reqwest 0.12.28",
@@ -697,6 +698,7 @@ dependencies = [
  "crank-adapter-rest",
  "crank-core",
  "crank-mapping",
+ "crank-metrics",
  "crank-observability",
  "crank-registry",
  "crank-runtime",
@@ -704,7 +706,7 @@ dependencies = [
  "crank-test-support",
  "crank-trace",
  "futures-util",
- "metrics",
+ "metrics-util",
  "opentelemetry",
  "opentelemetry_sdk",
  "reqwest 0.12.28",
@@ -758,11 +760,20 @@ dependencies = [
  "thiserror 2.0.18",
 ]
 
+[[package]]
+name = "crank-metrics"
+version = "0.3.1"
+dependencies = [
+ "metrics",
+ "metrics-util",
+]
+
 [[package]]
 name = "crank-observability"
 version = "0.3.1"
 dependencies = [
  "axum",
+ "crank-metrics",
  "metrics",
  "metrics-exporter-prometheus",
  "opentelemetry",
@@ -816,11 +827,13 @@ dependencies = [
  "crank-adapter-rest",
  "crank-core",
  "crank-mapping",
+ "crank-metrics",
  "crank-schema",
  "crank-trace",
  "futures-util",
  "hkdf 0.12.4",
  "metrics",
+ "metrics-util",
  "redis",
  "serde",
  "serde_json",
@@ -1075,6 +1088,12 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "endian-type"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d"
+
 [[package]]
 name = "equivalent"
 version = "1.0.2"
@@ -2009,6 +2028,7 @@ dependencies = [
  "crank-test-support",
  "futures-util",
  "metrics",
+ "metrics-util",
  "opentelemetry",
  "opentelemetry-proto",
  "opentelemetry_sdk",
@@ -2075,11 +2095,15 @@ version = "0.20.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "96f8722f8562635f92f8ed992f26df0532266eb03d5202607c20c0d7e9745e13"
 dependencies = [
+ "aho-corasick",
  "crossbeam-epoch",
  "crossbeam-utils",
  "hashbrown 0.16.1",
+ "indexmap 2.14.0",
  "metrics",
+ "ordered-float",
  "quanta",
+ "radix_trie",
  "rand 0.9.4",
  "rand_xoshiro",
  "rapidhash",
@@ -2112,6 +2136,15 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "nibble_vec"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43"
+dependencies = [
+ "smallvec",
+]
+
 [[package]]
 name = "nu-ansi-term"
 version = "0.50.3"
@@ -2296,6 +2329,15 @@ dependencies = [
  "thiserror 2.0.18",
 ]
 
+[[package]]
+name = "ordered-float"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "parking"
 version = "2.2.1"
@@ -2591,6 +2633,16 @@ version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
+[[package]]
+name = "radix_trie"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd"
+dependencies = [
+ "endian-type",
+ "nibble_vec",
+]
+
 [[package]]
 name = "rand"
 version = "0.9.4"

Cargo.toml

@@ -8,6 +8,7 @@ members = [
     "crates/crank-import",
     "crates/crank-schema",
     "crates/crank-mapping",
+    "crates/crank-metrics",
     "crates/crank-observability",
     "crates/crank-registry",
     "crates/crank-runtime",

apps/admin-api/Cargo.toml

@@ -24,7 +24,6 @@ crank-registry = { path = "../../crates/crank-registry" }
 crank-runtime = { path = "../../crates/crank-runtime" }
 crank-schema = { path = "../../crates/crank-schema" }
 crank-trace = { path = "../../crates/crank-trace" }
-metrics.workspace = true
 rand.workspace = true
 serde.workspace = true
 serde_json.workspace = true
@@ -41,6 +40,8 @@ uuid.workspace = true
 [dev-dependencies]
 async-trait = "0.1"
 crank-test-support = { path = "../../crates/crank-test-support" }
+metrics.workspace = true
+metrics-util = "0.20.4"
 opentelemetry.workspace = true
 opentelemetry_sdk.workspace = true
 reqwest.workspace = true

apps/admin-api/src/lib.rs

@@ -3,6 +3,7 @@ pub mod auth;
 pub mod dto;
 pub mod error;
 pub mod import_guidance;
+pub mod pool_metrics;
 pub mod rate_limit;
 pub mod request_context;
 pub mod routes;

apps/admin-api/src/main.rs

@@ -3,6 +3,7 @@ use std::{env, net::SocketAddr, path::PathBuf, time::Duration};
 use admin_api::{
     app::build_app,
     auth::{AuthSettings, BootstrapAdminConfig},
+    pool_metrics::spawn_postgres_pool_metrics,
     service::AdminServiceBuilder,
     state::AppState,
 };
@@ -16,10 +17,12 @@ use crank_runtime::{
     RequestRateLimitConfig, RequestRateLimiter, RuntimeCacheConfig, RuntimeCacheStores,
     RuntimeLimits, SecretCrypto,
 };
-use sqlx::{PgPool, postgres::PgConnectOptions};
+use sqlx::postgres::PgConnectOptions;
 use tokio::net::TcpListener;
 use tracing::{info, warn};
 
+const MAX_INVOCATION_LOG_RETENTION_DAYS: i64 = 36_500;
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let observability = crank_observability::init(ObservabilityConfig::from_env(
@@ -105,8 +108,7 @@ async fn run(
     .with_outbound_http_policy(outbound_http_policy)
     .with_identity_provider(std::sync::Arc::new(identity_provider))
     .build();
-    let invocation_log_retention_days =
-        positive_i64_from_env("CRANK_INVOCATION_LOG_RETENTION_DAYS", 30)?;
+    let invocation_log_retention_days = invocation_log_retention_days_from_env()?;
     service.bootstrap_admin_user().await?;
     if env_flag("CRANK_DEMO_SEED") {
         service.seed_demo_assets().await?;
@@ -158,17 +160,17 @@ async fn run(
     Ok(())
 }
 
-fn positive_i64_from_env(
-    name: &'static str,
-    default: i64,
-) -> Result<i64, Box<dyn std::error::Error>> {
-    let value = match env::var(name) {
+fn invocation_log_retention_days_from_env() -> Result<i64, Box<dyn std::error::Error>> {
+    const NAME: &str = "CRANK_INVOCATION_LOG_RETENTION_DAYS";
+    let value = match env::var(NAME) {
         Ok(raw) => raw.parse::<i64>()?,
-        Err(env::VarError::NotPresent) => default,
+        Err(env::VarError::NotPresent) => 30,
         Err(error) => return Err(error.into()),
     };
-    if value <= 0 {
-        return Err(format!("{name} must be greater than zero").into());
+    if !(1..=MAX_INVOCATION_LOG_RETENTION_DAYS).contains(&value) {
+        return Err(
+            format!("{NAME} must be between 1 and {MAX_INVOCATION_LOG_RETENTION_DAYS}").into(),
+        );
     }
     Ok(value)
 }
@@ -196,16 +198,6 @@ fn spawn_invocation_log_cleanup(service: admin_api::service::AdminService, reten
     });
 }
 
-fn spawn_postgres_pool_metrics(pool: PgPool) {
-    tokio::spawn(async move {
-        let mut interval = tokio::time::interval(Duration::from_secs(5));
-        loop {
-            interval.tick().await;
-            crank_observability::record_db_pool_connections(pool.size(), pool.num_idle());
-        }
-    });
-}
-
 fn env_flag(name: &str) -> bool {
     matches!(
         env::var(name)

apps/admin-api/src/pool_metrics.rs

@@ -0,0 +1,13 @@
+use std::time::Duration;
+
+use sqlx::PgPool;
+
+pub fn spawn_postgres_pool_metrics(pool: PgPool) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_secs(5));
+        loop {
+            interval.tick().await;
+            crank_observability::record_db_pool_connections(pool.size(), pool.num_idle());
+        }
+    })
+}

apps/admin-api/src/request_context.rs

@@ -1,5 +1,5 @@
 use axum::{
-    extract::Request,
+    extract::{MatchedPath, Request},
     http::{HeaderName, HeaderValue},
     middleware::Next,
     response::Response,
@@ -19,7 +19,11 @@ pub async fn apply_request_context(mut request: Request, next: Next) -> Response
         request_id: RequestId::resolve_from_headers(request.headers()).into_string(),
     };
     let method = request.method().clone();
-    let path = request.uri().path().to_owned();
+    let route = request
+        .extensions()
+        .get::<MatchedPath>()
+        .map_or("unmatched", MatchedPath::as_str)
+        .to_owned();
     let span = info_span!(
         target: "crank::trace",
         "http.request",
@@ -34,7 +38,7 @@ pub async fn apply_request_context(mut request: Request, next: Next) -> Response
             name: "admin.request.completed",
             request_id = %context.request_id,
             method = %method,
-            path,
+            route,
             status = response.status().as_u16(),
             "admin request completed"
         );

apps/admin-api/tests/integration/request_context.rs

@@ -65,6 +65,7 @@ async fn logs_request_completion_and_rejects_untrusted_values() {
         .unwrap();
     assert_eq!(event["request_id"], "req_admin_trace_123");
     assert_eq!(event["fields"]["status"], 200);
+    assert_eq!(event["fields"]["route"], "/probe");
 
     let invalid_response = app
         .oneshot(

apps/admin-api/tests/product_metrics.rs

@@ -0,0 +1,89 @@
+#[path = "integration/common.rs"]
+mod common;
+
+use std::{collections::BTreeSet, time::Duration};
+
+use common::*;
+use metrics_util::debugging::DebuggingRecorder;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn real_admin_routes_and_postgres_sampler_emit_bounded_metrics() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    recorder
+        .install()
+        .expect("isolated integration test recorder");
+
+    let registry = test_registry().await;
+    let sampler = admin_api::pool_metrics::spawn_postgres_pool_metrics(registry.pool().clone());
+    let server = spawn_admin_api(build_test_app(
+        registry,
+        test_storage_root("product_metrics"),
+    ))
+    .await;
+    let client = authorized_client(&server).await;
+    client
+        .get(format!("{server}/operations"))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+
+    let snapshot = wait_for_metrics(&snapshotter).await;
+    assert!(snapshot.iter().any(|(key, _, _, _)| {
+        key.key().name() == "crank_http_requests_total"
+            && has_label(
+                key.key(),
+                "route",
+                "/api/admin/workspaces/{workspace_id}/operations",
+            )
+            && has_label(key.key(), "method", "GET")
+            && has_label(key.key(), "status_class", "2xx")
+    }));
+    let pool_states = snapshot
+        .iter()
+        .filter(|(key, _, _, _)| key.key().name() == "crank_db_pool_connections")
+        .filter_map(|(key, _, _, _)| {
+            key.key()
+                .labels()
+                .find(|label| label.key() == "state")
+                .map(|label| label.value().to_owned())
+        })
+        .collect::<BTreeSet<_>>();
+    assert_eq!(
+        pool_states,
+        BTreeSet::from(["idle".to_owned(), "used".to_owned()])
+    );
+
+    sampler.abort();
+}
+
+async fn wait_for_metrics(
+    snapshotter: &metrics_util::debugging::Snapshotter,
+) -> Vec<(
+    metrics_util::CompositeKey,
+    Option<metrics::Unit>,
+    Option<metrics::SharedString>,
+    metrics_util::debugging::DebugValue,
+)> {
+    for _ in 0..50 {
+        let snapshot = snapshotter.snapshot().into_vec();
+        let has_http = snapshot
+            .iter()
+            .any(|(key, _, _, _)| key.key().name() == "crank_http_requests_total");
+        let has_pool = snapshot
+            .iter()
+            .any(|(key, _, _, _)| key.key().name() == "crank_db_pool_connections");
+        if has_http && has_pool {
+            return snapshot;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("product metrics did not become visible");
+}
+
+fn has_label(key: &metrics::Key, name: &str, value: &str) -> bool {
+    key.labels()
+        .any(|label| label.key() == name && label.value() == value)
+}

apps/mcp-server/Cargo.toml

@@ -21,7 +21,6 @@ crank-registry = { path = "../../crates/crank-registry" }
 crank-runtime = { path = "../../crates/crank-runtime" }
 crank-schema = { path = "../../crates/crank-schema" }
 futures-util = "0.3"
-metrics.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 sha2.workspace = true
@@ -37,6 +36,7 @@ uuid.workspace = true
 crank-mapping = { path = "../../crates/crank-mapping" }
 crank-schema = { path = "../../crates/crank-schema" }
 crank-test-support = { path = "../../crates/crank-test-support" }
+metrics.workspace = true
 opentelemetry.workspace = true
 opentelemetry-proto.workspace = true
 opentelemetry_sdk.workspace = true
@@ -44,3 +44,4 @@ prost.workspace = true
 reqwest.workspace = true
 tower.workspace = true
 tracing-opentelemetry.workspace = true
+metrics-util = "0.20.4"

apps/mcp-server/src/lib.rs

@@ -0,0 +1 @@
+pub mod pool_metrics;

apps/mcp-server/src/main.rs

@@ -13,7 +13,8 @@ use crank_runtime::{
     RequestRateLimitConfig, RequestRateLimiter, RuntimeCacheConfig, RuntimeCacheStores,
     RuntimeLimits, SecretCrypto,
 };
-use sqlx::{PgPool, postgres::PgConnectOptions};
+use mcp_server::pool_metrics::spawn_postgres_pool_metrics;
+use sqlx::postgres::PgConnectOptions;
 use tokio::net::TcpListener;
 use tracing::info;
 
@@ -162,13 +163,3 @@ fn mcp_api_rate_limit_config_from_env() -> Result<RequestRateLimitConfig, Box<dy
 
     Ok(RequestRateLimitConfig::new(requests_per_second, burst)?)
 }
-
-fn spawn_postgres_pool_metrics(pool: PgPool) {
-    tokio::spawn(async move {
-        let mut interval = tokio::time::interval(Duration::from_secs(5));
-        loop {
-            interval.tick().await;
-            crank_observability::record_db_pool_connections(pool.size(), pool.num_idle());
-        }
-    });
-}

apps/mcp-server/src/pool_metrics.rs

@@ -0,0 +1,13 @@
+use std::time::Duration;
+
+use sqlx::PgPool;
+
+pub fn spawn_postgres_pool_metrics(pool: PgPool) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_secs(5));
+        loop {
+            interval.tick().await;
+            crank_observability::record_db_pool_connections(pool.size(), pool.num_idle());
+        }
+    })
+}

apps/mcp-server/tests/product_metrics.rs

@@ -0,0 +1,345 @@
+#[path = "integration/common.rs"]
+mod common;
+
+use std::{collections::BTreeSet, time::Duration};
+
+use common::*;
+use crank_core::PlatformApiKeyScope;
+use crank_registry::PublishRequest;
+use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter};
+use serde_json::json;
+use time::{OffsetDateTime, format_description::well_known::Rfc3339};
+
+#[tokio::test]
+async fn real_mcp_runtime_upstream_postgres_and_catalog_paths_emit_bounded_metrics() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    recorder
+        .install()
+        .expect("isolated integration test recorder");
+
+    let registry = test_registry().await;
+    let pool_sampler =
+        mcp_server::pool_metrics::spawn_postgres_pool_metrics(registry.pool().clone());
+    let upstream_base_url = spawn_upstream_server().await;
+    let operation = test_operation(&upstream_base_url, "crm_create_lead");
+    registry
+        .create_operation(&test_workspace_id(), &operation, Some("metrics-test"))
+        .await
+        .unwrap();
+    registry
+        .publish_operation(PublishRequest {
+            workspace_id: &test_workspace_id(),
+            operation_id: &operation.id,
+            version: 1,
+            published_at: &OffsetDateTime::parse("2026-07-31T00:00:00Z", &Rfc3339).unwrap(),
+            published_by: Some("metrics-test"),
+        })
+        .await
+        .unwrap();
+    publish_agent_for_operation(&registry, &operation, "metrics-agent").await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "metrics-agent",
+        "metrics-client",
+        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
+    )
+    .await;
+    let mut varied_products = Vec::new();
+    for index in 0..3 {
+        let upstream_url = spawn_upstream_server().await;
+        let operation_name = format!("customer_operation_{index}");
+        let operation = test_operation(&upstream_url, &operation_name);
+        registry
+            .create_operation(&test_workspace_id(), &operation, Some("metrics-test"))
+            .await
+            .unwrap();
+        registry
+            .publish_operation(PublishRequest {
+                workspace_id: &test_workspace_id(),
+                operation_id: &operation.id,
+                version: 1,
+                published_at: &OffsetDateTime::parse("2026-07-31T00:00:00Z", &Rfc3339).unwrap(),
+                published_by: Some("metrics-test"),
+            })
+            .await
+            .unwrap();
+        let agent_slug = format!("customer-agent-{index}");
+        publish_agent_for_operation(&registry, &operation, &agent_slug).await;
+        let key = create_platform_api_key(
+            &registry,
+            &agent_slug,
+            &format!("metrics-client-{index}"),
+            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
+        )
+        .await;
+        varied_products.push((agent_slug, key, operation_name, upstream_url));
+    }
+
+    let base_url = spawn_mcp_server(build_test_app(
+        registry.clone(),
+        Duration::ZERO,
+        Some("https://crank.example.com".to_owned()),
+    ))
+    .await;
+    let client = reqwest::Client::new();
+    let mcp_url = agent_mcp_url(&base_url, "metrics-agent");
+    let malformed = client
+        .post(&mcp_url)
+        .header(reqwest::header::ACCEPT, "application/json")
+        .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header(reqwest::header::CONTENT_TYPE, "application/json")
+        .body("{")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(malformed.status(), reqwest::StatusCode::BAD_REQUEST);
+
+    let invalid_session_header = client
+        .post(&mcp_url)
+        .header(
+            reqwest::header::ACCEPT,
+            "application/json, text/event-stream",
+        )
+        .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header(
+            "MCP-Session-Id",
+            reqwest::header::HeaderValue::from_bytes(b"\xff").unwrap(),
+        )
+        .json(&json!({
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {"protocolVersion": "2025-11-25", "capabilities": {}}
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        invalid_session_header.status(),
+        reqwest::StatusCode::BAD_REQUEST
+    );
+
+    let session_id = initialize_session(&client, &mcp_url, &api_key).await;
+    assert_gauge(&snapshotter, "crank_mcp_active_sessions", 1.0).await;
+    let _ = post_jsonrpc(
+        &client,
+        &mcp_url,
+        &api_key,
+        Some(&session_id),
+        json!({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}),
+    )
+    .await;
+    call_tool(
+        &client,
+        &mcp_url,
+        &api_key,
+        &session_id,
+        "crm_create_lead",
+        "first@example.com",
+        "req-first",
+    )
+    .await;
+    delete_session(&client, &mcp_url, &api_key, &session_id).await;
+    assert_gauge(&snapshotter, "crank_mcp_active_sessions", 0.0).await;
+    let session_id = initialize_session(&client, &mcp_url, &api_key).await;
+    assert_gauge(&snapshotter, "crank_mcp_active_sessions", 1.0).await;
+    for (index, (agent_slug, key, operation_name, _)) in varied_products.iter().enumerate() {
+        let url = agent_mcp_url(&base_url, agent_slug);
+        let session = initialize_session(&client, &url, key).await;
+        call_tool(
+            &client,
+            &url,
+            key,
+            &session,
+            operation_name,
+            &format!("variant-{index}@example.com"),
+            &format!("req-product-{index}"),
+        )
+        .await;
+        delete_session(&client, &url, key, &session).await;
+    }
+    send_invalid_workspace_request(&client, &base_url, &api_key, "customer-workspace-seed").await;
+    tokio::time::sleep(Duration::from_millis(25)).await;
+
+    let first_snapshot = snapshotter.snapshot().into_vec();
+    let first_series = series(&first_snapshot);
+    assert!(first_snapshot.iter().any(|(key, _, _, _)| {
+        key.key().name() == "crank_mcp_requests_total"
+            && has_label(key.key(), "method", "invalid")
+            && has_label(key.key(), "outcome", "client_error")
+    }));
+    let names = first_snapshot
+        .iter()
+        .map(|(key, _, _, _)| key.key().name())
+        .collect::<BTreeSet<_>>();
+    for expected in [
+        "crank_mcp_requests_total",
+        "crank_mcp_active_sessions",
+        "crank_tool_invocations_total",
+        "crank_tool_invocation_duration_seconds",
+        "crank_upstream_requests_total",
+        "crank_upstream_request_duration_seconds",
+        "crank_db_pool_connections",
+        "crank_catalog_tools",
+        "crank_catalog_estimated_context_tokens",
+        "crank_catalog_warnings",
+    ] {
+        assert!(
+            names.contains(expected),
+            "missing product metric {expected}"
+        );
+    }
+
+    for index in 0..32 {
+        call_tool(
+            &client,
+            &mcp_url,
+            &api_key,
+            &session_id,
+            "crm_create_lead",
+            &format!("user-{index}@example.com"),
+            &format!("req-metrics-{index}"),
+        )
+        .await;
+        send_invalid_workspace_request(
+            &client,
+            &base_url,
+            &api_key,
+            &format!("customer-workspace-{index}"),
+        )
+        .await;
+    }
+    tokio::time::sleep(Duration::from_millis(25)).await;
+
+    let diverse_snapshot = snapshotter.snapshot().into_vec();
+    assert_eq!(series(&diverse_snapshot), first_series);
+    delete_session(&client, &mcp_url, &api_key, &session_id).await;
+    assert_gauge(&snapshotter, "crank_mcp_active_sessions", 0.0).await;
+    let rendered = format!("{diverse_snapshot:?}");
+    let mut forbidden = vec![
+        "metrics-agent",
+        "crm_create_lead",
+        "user-31@example.com",
+        "req-metrics-31",
+        &upstream_base_url,
+    ];
+    for (agent_slug, _, operation_name, upstream_url) in &varied_products {
+        forbidden.extend([
+            agent_slug.as_str(),
+            operation_name.as_str(),
+            upstream_url.as_str(),
+        ]);
+    }
+    for forbidden in forbidden {
+        assert!(
+            !rendered.contains(forbidden),
+            "untrusted value leaked into metric labels: {forbidden}"
+        );
+    }
+    pool_sampler.abort();
+}
+
+async fn delete_session(client: &reqwest::Client, mcp_url: &str, api_key: &str, session_id: &str) {
+    let deleted = client
+        .delete(mcp_url)
+        .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", session_id)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(deleted.status(), reqwest::StatusCode::NO_CONTENT);
+}
+
+async fn call_tool(
+    client: &reqwest::Client,
+    mcp_url: &str,
+    api_key: &str,
+    session_id: &str,
+    tool_name: &str,
+    email: &str,
+    request_id: &str,
+) {
+    let response = post_jsonrpc_response(
+        client,
+        mcp_url,
+        api_key,
+        Some(session_id),
+        Some(request_id),
+        json!({
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "method": "tools/call",
+            "params": {
+                "name": tool_name,
+                "arguments": {"email": email}
+            }
+        }),
+    )
+    .await;
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+}
+
+async fn send_invalid_workspace_request(
+    client: &reqwest::Client,
+    base_url: &str,
+    api_key: &str,
+    workspace_slug: &str,
+) {
+    let response = client
+        .post(format!("{base_url}/v1/{workspace_slug}/metrics-agent"))
+        .header(reqwest::header::ACCEPT, "application/json")
+        .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .json(&json!({
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {"protocolVersion": "2025-11-25", "capabilities": {}}
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert!(response.status().is_client_error());
+}
+
+fn series(
+    snapshot: &[(
+        metrics_util::CompositeKey,
+        Option<metrics::Unit>,
+        Option<metrics::SharedString>,
+        metrics_util::debugging::DebugValue,
+    )],
+) -> BTreeSet<String> {
+    snapshot
+        .iter()
+        .map(|(key, _, _, _)| format!("{key:?}"))
+        .collect()
+}
+
+fn has_label(key: &metrics::Key, name: &str, value: &str) -> bool {
+    key.labels()
+        .any(|label| label.key() == name && label.value() == value)
+}
+
+async fn assert_gauge(snapshotter: &Snapshotter, name: &str, expected: f64) {
+    for _ in 0..50 {
+        let value = snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .find_map(|(key, _, _, value)| {
+                if key.key().name() != name {
+                    return None;
+                }
+                match value {
+                    DebugValue::Gauge(value) => Some(value.into_inner()),
+                    _ => None,
+                }
+            });
+        if value == Some(expected) {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("{name} did not become {expected}");
+}

crates/crank-adapter-rest/Cargo.toml

@@ -9,9 +9,9 @@ version.workspace = true
 [dependencies]
 async-trait = "0.1"
 crank-core = { path = "../crank-core" }
+crank-metrics = { path = "../crank-metrics" }
 crank-trace = { path = "../crank-trace" }
 futures-util = "0.3"
-metrics.workspace = true
 opentelemetry.workspace = true
 reqwest = { workspace = true, features = ["stream"] }
 serde.workspace = true

crates/crank-adapter-rest/src/client.rs

@@ -7,6 +7,7 @@ use std::{
 };
 
 use crank_core::{HttpMethod, RestTarget};
+use crank_metrics::{UpstreamOperationKind, UpstreamOutcome, UpstreamRequestMetrics};
 use crank_trace::{ErrorCategory, Stage, StageOutcome};
 use futures_util::StreamExt;
 use opentelemetry::{global, propagation::Injector, trace::TraceContextExt};
@@ -71,24 +72,13 @@ impl RestAdapter {
         target: &RestTarget,
         request: &RestRequest,
     ) -> Result<RestResponse, RestAdapterError> {
-        let started_at = std::time::Instant::now();
+        let request_metrics = UpstreamRequestMetrics::start(UpstreamOperationKind::Rest);
         let result = self.execute_inner(target, request).await;
         let outcome = match &result {
-            Ok(_) => "success",
+            Ok(_) => UpstreamOutcome::Success,
             Err(error) => upstream_outcome(error),
         };
-        metrics::counter!(
-            "crank_upstream_requests_total",
-            "operation_kind" => "rest",
-            "outcome" => outcome
-        )
-        .increment(1);
-        metrics::histogram!(
-            "crank_upstream_request_duration_seconds",
-            "operation_kind" => "rest",
-            "outcome" => outcome
-        )
-        .record(started_at.elapsed().as_secs_f64());
+        request_metrics.complete(outcome);
         result
     }
 
@@ -149,27 +139,27 @@ impl RestAdapter {
     }
 }
 
-fn upstream_outcome(error: &RestAdapterError) -> &'static str {
+fn upstream_outcome(error: &RestAdapterError) -> UpstreamOutcome {
     match error {
         RestAdapterError::UnexpectedStatus { status, .. } if (400..500).contains(status) => {
-            "client_error"
+            UpstreamOutcome::ClientError
         }
         RestAdapterError::UnexpectedStatus { status, .. } if (500..600).contains(status) => {
-            "server_error"
+            UpstreamOutcome::ServerError
         }
-        RestAdapterError::UnexpectedStatus { .. } => "unexpected_status",
-        RestAdapterError::Transport(error) if error.is_timeout() => "timeout",
-        RestAdapterError::Transport(_) => "transport_error",
-        RestAdapterError::ResponseTooLarge { .. } => "response_too_large",
-        RestAdapterError::TargetNotAllowed { .. } => "rejected",
-        RestAdapterError::WindowExpired => "window_expired",
-        RestAdapterError::InvalidSseEvent => "invalid_response",
+        RestAdapterError::UnexpectedStatus { .. } => UpstreamOutcome::UnexpectedStatus,
+        RestAdapterError::Transport(error) if error.is_timeout() => UpstreamOutcome::Timeout,
+        RestAdapterError::Transport(_) => UpstreamOutcome::TransportError,
+        RestAdapterError::ResponseTooLarge { .. } => UpstreamOutcome::ResponseTooLarge,
+        RestAdapterError::TargetNotAllowed { .. } => UpstreamOutcome::Rejected,
+        RestAdapterError::WindowExpired => UpstreamOutcome::WindowExpired,
+        RestAdapterError::InvalidSseEvent => UpstreamOutcome::InvalidResponse,
         RestAdapterError::InvalidBaseUrl { .. }
         | RestAdapterError::InvalidPathParameter { .. }
         | RestAdapterError::InvalidQueryParameter { .. }
         | RestAdapterError::InvalidHeaderName { .. }
-        | RestAdapterError::InvalidHeaderValue { .. } => "invalid_request",
-        RestAdapterError::InvalidConfiguration { .. } => "configuration",
+        | RestAdapterError::InvalidHeaderValue { .. } => UpstreamOutcome::InvalidRequest,
+        RestAdapterError::InvalidConfiguration { .. } => UpstreamOutcome::Configuration,
     }
 }
 

crates/crank-community-mcp/Cargo.toml

@@ -12,13 +12,13 @@ axum.workspace = true
 base64.workspace = true
 crank-adapter-rest = { path = "../crank-adapter-rest" }
 crank-core = { path = "../crank-core" }
+crank-metrics = { path = "../crank-metrics" }
 crank-observability = { path = "../crank-observability" }
 crank-registry = { path = "../crank-registry" }
 crank-runtime = { path = "../crank-runtime" }
 crank-schema = { path = "../crank-schema" }
 crank-trace = { path = "../crank-trace" }
 futures-util = "0.3"
-metrics.workspace = true
 reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
@@ -33,6 +33,7 @@ uuid.workspace = true
 [dev-dependencies]
 crank-mapping = { path = "../crank-mapping" }
 crank-test-support = { path = "../crank-test-support" }
+metrics-util = "0.20.4"
 opentelemetry.workspace = true
 opentelemetry_sdk.workspace = true
 tracing-opentelemetry.workspace = true

crates/crank-community-mcp/src/app.rs

@@ -7,7 +7,7 @@ use std::{
 
 use axum::{
     Json, Router,
-    extract::{Extension, Path, State},
+    extract::{Extension, Path, State, rejection::JsonRejection},
     http::{HeaderMap, StatusCode},
     response::{IntoResponse, Response, sse::Event},
     routing::{get, post},
@@ -31,7 +31,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 use time::OffsetDateTime;
 use tokio::sync::Semaphore;
-use tracing::{Instrument, info, warn};
+use tracing::{Instrument, info};
 
 use crate::{
     access::{
@@ -48,23 +48,23 @@ use crate::{
     manifest::catalog_tool_definitions,
     rate_limit::{rate_limited_jsonrpc_response, rate_limited_status_response},
     request_context::{RequestContext, apply_request_context},
-    session::{SessionState, SharedSessionStore},
+    session::{ActiveSessionMetrics, SessionState, SharedSessionStore, spawn_session_cleanup},
     tool_error::{
         ToolErrorContract, generic_tool_error_contract, runtime_error_code,
         tool_error_contract_from_runtime, tool_error_text, tool_error_value,
     },
     tool_search::handle_catalog_tool_call,
     transport::{
-        AllowedOrigins, HEADER_MCP_SESSION_ID, ResponseMode, json_response,
-        negotiate_post_response_mode, protocol_version_from_headers, session_id_from_headers,
-        sse_response, transport_response, validate_get_accept_header, validate_origin,
-        validate_session_protocol_version, with_request_id_header,
+        AllowedOrigins, ResponseMode, json_response, negotiate_post_response_mode,
+        protocol_version_from_headers, session_id_from_headers, sse_response, transport_response,
+        validate_get_accept_header, validate_origin, validate_session_protocol_version,
+        with_request_id_header,
     },
 };
 mod invocation_history;
 mod metrics;
 mod stages;
-use self::metrics::{ActiveSessionGuard, McpRequestMetrics};
+use self::metrics::{ActiveStreamGuard, McpRequestMetrics};
 use self::stages::{
     enforce_traced_rate_limit, require_traced_approval_access, require_traced_machine_access,
 };
@@ -84,6 +84,7 @@ pub(super) struct AppState {
     pub(super) api_rate_limiter: RequestRateLimiter,
     secret_crypto: SecretCrypto,
     sessions: SharedSessionStore,
+    session_metrics: ActiveSessionMetrics,
     session_slots: Arc<Semaphore>,
     pub(super) credential_verifier: SharedMachineCredentialVerifier,
     allowed_origins: AllowedOrigins,
@@ -230,6 +231,7 @@ fn build_app_inner(
     max_concurrent_sessions: usize,
     start_background_workers: bool,
 ) -> Router {
+    let session_metrics = ActiveSessionMetrics::start(Arc::clone(&sessions));
     let state = Arc::new(AppState {
         registry: registry.clone(),
         catalog: PublishedToolCatalog::new(registry, refresh_interval, coordination_store),
@@ -237,13 +239,18 @@ fn build_app_inner(
         api_rate_limiter,
         secret_crypto,
         sessions,
+        session_metrics,
         session_slots: Arc::new(Semaphore::new(max_concurrent_sessions)),
         credential_verifier,
         allowed_origins: AllowedOrigins::new(public_base_url),
     });
     if start_background_workers {
         spawn_approval_recovery(Arc::clone(&state));
-        spawn_session_cleanup(Arc::clone(&state));
+        spawn_session_cleanup(
+            Arc::clone(&state.sessions),
+            state.session_metrics.clone(),
+            SESSION_CLEANUP_INTERVAL,
+        );
     }
 
     Router::new()
@@ -283,28 +290,6 @@ async fn health() -> Json<Value> {
     }))
 }
 
-fn spawn_session_cleanup(state: Arc<AppState>) {
-    tokio::spawn(async move {
-        let mut interval = tokio::time::interval(SESSION_CLEANUP_INTERVAL);
-        loop {
-            interval.tick().await;
-            match state
-                .sessions
-                .cleanup_expired(OffsetDateTime::now_utc())
-                .await
-            {
-                Ok(removed) if removed > 0 => {
-                    info!(name: "mcp.session_cleanup.completed", removed);
-                }
-                Ok(_) => {}
-                Err(_) => {
-                    warn!(name: "mcp.session_cleanup.failed", error_category = "session_store");
-                }
-            }
-        }
-    });
-}
-
 async fn readiness(State(state): State<Arc<AppState>>) -> Response {
     match state.registry.ping().await {
         Ok(()) => Json(json!({
@@ -664,7 +649,7 @@ async fn mcp_get(
         return status.into_response();
     }
 
-    let Ok(permit) = ActiveSessionGuard::try_acquire(&state.session_slots) else {
+    let Ok(permit) = ActiveStreamGuard::try_acquire(&state.session_slots) else {
         return StatusCode::TOO_MANY_REQUESTS.into_response();
     };
 
@@ -704,7 +689,10 @@ async fn mcp_delete(
                     && session.agent_slug == path.agent_slug =>
             {
                 match state.sessions.delete(&session_id).await {
-                    Ok(true) => StatusCode::NO_CONTENT.into_response(),
+                    Ok(true) => {
+                        state.session_metrics.refresh();
+                        StatusCode::NO_CONTENT.into_response()
+                    }
                     Ok(false) => StatusCode::NOT_FOUND.into_response(),
                     Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
                 }
@@ -723,8 +711,17 @@ async fn mcp_post(
     State(state): State<Arc<AppState>>,
     Extension(request_context): Extension<RequestContext>,
     headers: HeaderMap,
-    Json(message): Json<Value>,
+    payload: Result<Json<Value>, JsonRejection>,
 ) -> Response {
+    let message = match payload {
+        Ok(Json(message)) => message,
+        Err(rejection) => {
+            let mut request_metrics = McpRequestMetrics::invalid();
+            let response = rejection.into_response();
+            request_metrics.complete(&response);
+            return with_request_id_header(response, &request_context.request_id);
+        }
+    };
     let mut request_metrics = McpRequestMetrics::new(&message);
     let transport_request_id = request_context.request_id;
     info!(
@@ -736,78 +733,77 @@ async fn mcp_post(
         "mcp request received"
     );
 
-    if let Err(status) = validate_origin(&state.allowed_origins, &headers) {
-        return with_request_id_header(status.into_response(), &transport_request_id);
+    let response = mcp_post_response(&path, state, &headers, &message, &transport_request_id).await;
+    request_metrics.complete(&response);
+    with_request_id_header(response, &transport_request_id)
+}
+
+async fn mcp_post_response(
+    path: &AgentRoutePath,
+    state: Arc<AppState>,
+    headers: &HeaderMap,
+    message: &Value,
+    transport_request_id: &str,
+) -> Response {
+    if let Err(status) = validate_origin(&state.allowed_origins, headers) {
+        return status.into_response();
     }
 
-    let response_mode = match negotiate_post_response_mode(&headers) {
-        Ok(mode) => request_metrics.set_response_mode(mode),
-        Err(status) => {
-            return with_request_id_header(status.into_response(), &transport_request_id);
-        }
+    let response_mode = match negotiate_post_response_mode(headers) {
+        Ok(mode) => mode,
+        Err(status) => return status.into_response(),
     };
 
-    if is_response(&message) || is_notification(&message) && method_name(&message).is_none() {
-        return with_request_id_header(StatusCode::ACCEPTED.into_response(), &transport_request_id);
+    let request_session_id = match session_id_from_headers(headers) {
+        Ok(session_id) => session_id,
+        Err(status) => return status.into_response(),
+    };
+
+    if is_response(message) || is_notification(message) && method_name(message).is_none() {
+        return StatusCode::ACCEPTED.into_response();
     }
 
-    let protocol_version = match protocol_version_from_headers(&headers) {
+    let protocol_version = match protocol_version_from_headers(headers) {
         Ok(value) => value,
-        Err(status) => {
-            return with_request_id_header(status.into_response(), &transport_request_id);
-        }
+        Err(status) => return status.into_response(),
     };
 
-    let rate_limit_result = enforce_traced_rate_limit(&state, &path, &headers).await;
+    let rate_limit_result = enforce_traced_rate_limit(&state, path, headers).await;
     if let Err(error) = rate_limit_result {
-        return with_request_id_header(
-            rate_limited_jsonrpc_response(&message, response_mode, &protocol_version, error),
-            &transport_request_id,
-        );
+        return rate_limited_jsonrpc_response(message, response_mode, &protocol_version, error);
     }
-    if let Some(session_id) = headers.get(HEADER_MCP_SESSION_ID)
-        && let Ok(session_id) = session_id.to_str()
-    {
+    if let Some(session_id) = request_session_id.as_deref() {
         let session = match state.sessions.get(session_id).await {
             Ok(session) => session,
-            Err(_) => {
-                return with_request_id_header(
-                    StatusCode::INTERNAL_SERVER_ERROR.into_response(),
-                    &transport_request_id,
-                );
-            }
+            Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(),
         };
         if let Some(session) = session
             && let Err(status) =
-                validate_session_protocol_version(&headers, &session.protocol_version)
+                validate_session_protocol_version(headers, &session.protocol_version)
         {
-            return with_request_id_header(status.into_response(), &transport_request_id);
+            return status.into_response();
         }
     }
 
-    let required_scope = match method_name(&message) {
+    let required_scope = match method_name(message) {
         Some("tools/call") => PlatformApiKeyScope::Write,
         _ => PlatformApiKeyScope::Read,
     };
-    let access_result =
-        require_traced_machine_access(&state, &path, &headers, required_scope).await;
+    let access_result = require_traced_machine_access(&state, path, headers, required_scope).await;
     let credential = match access_result {
         Ok(credential) => credential,
-        Err(error) => {
-            return with_request_id_header(error.into_response(), &transport_request_id);
-        }
+        Err(error) => return error.into_response(),
     };
 
-    let response = match method_name(&message) {
-        Some("initialize") if is_request(&message) => {
-            handle_initialize(state, &path, &message, response_mode).await
+    match method_name(message) {
+        Some("initialize") if is_request(message) => {
+            handle_initialize(state, path, message, response_mode).await
         }
-        Some("notifications/initialized") if is_notification(&message) => {
-            handle_initialized_notification(state, &path, &headers).await
+        Some("notifications/initialized") if is_notification(message) => {
+            handle_initialized_notification(state, path, headers).await
         }
-        Some("ping") if is_request(&message) => {
-            let session = match require_initialized_session(&state, &path, &headers, &message).await
-            {
+        Some("ping") if is_request(message) => {
+            let session = match require_initialized_session(&state, path, headers, message).await {
                 Ok(session) => session,
                 Err(response) => return response,
             };
@@ -815,7 +811,7 @@ async fn mcp_post(
             transport_response(
                 StatusCode::OK,
                 jsonrpc_result(
-                    request_id(&message),
+                    request_id(message),
                     json!({ "protocolVersion": session.protocol_version }),
                 ),
                 response_mode,
@@ -823,9 +819,8 @@ async fn mcp_post(
                 Some(&session.protocol_version),
             )
         }
-        Some("tools/list") if is_request(&message) => {
-            let session = match require_initialized_session(&state, &path, &headers, &message).await
-            {
+        Some("tools/list") if is_request(message) => {
+            let session = match require_initialized_session(&state, path, headers, message).await {
                 Ok(session) => session,
                 Err(response) => return response,
             };
@@ -840,27 +835,26 @@ async fn mcp_post(
 
                     transport_response(
                         StatusCode::OK,
-                        jsonrpc_result(request_id(&message), json!({ "tools": definitions })),
+                        jsonrpc_result(request_id(message), json!({ "tools": definitions })),
                         response_mode,
                         None,
                         Some(&session.protocol_version),
                     )
                 }
-                Err(error) => internal_jsonrpc_error(&message, error),
+                Err(error) => internal_jsonrpc_error(message, error),
             }
         }
-        Some("tools/call") if is_request(&message) => {
-            let session = match require_initialized_session(&state, &path, &headers, &message).await
-            {
+        Some("tools/call") if is_request(message) => {
+            let session = match require_initialized_session(&state, path, headers, message).await {
                 Ok(session) => session,
                 Err(response) => return response,
             };
-            let tool_call_params: ToolCallParams = match serde_json::from_value(params(&message)) {
+            let tool_call_params: ToolCallParams = match serde_json::from_value(params(message)) {
                 Ok(value) => value,
                 Err(error) => {
                     return transport_response(
                         StatusCode::OK,
-                        jsonrpc_error(request_id(&message), -32602, error.to_string()),
+                        jsonrpc_error(request_id(message), -32602, error.to_string()),
                         response_mode,
                         None,
                         Some(&session.protocol_version),
@@ -883,27 +877,27 @@ async fn mcp_post(
                     handle_catalog_tool_call(
                         state.clone(),
                         &session,
-                        &message,
+                        message,
                         response_mode,
                         &credential,
                         &catalog,
                         &tool_call_params.name,
                         arguments,
-                        &transport_request_id,
+                        transport_request_id,
                     )
                     .await
                 }
-                Err(error) => internal_jsonrpc_error(&message, error),
+                Err(error) => internal_jsonrpc_error(message, error),
             }
         }
-        Some(method) if is_notification(&message) => {
+        Some(method) if is_notification(message) => {
             let _ = method;
             StatusCode::ACCEPTED.into_response()
         }
         Some(method) => transport_response(
             StatusCode::OK,
             jsonrpc_error(
-                request_id(&message),
+                request_id(message),
                 -32601,
                 format!("method {method} is not supported"),
             ),
@@ -918,10 +912,7 @@ async fn mcp_post(
             None,
             Some(&protocol_version),
         ),
-    };
-
-    request_metrics.complete(response.status());
-    with_request_id_header(response, &transport_request_id)
+    }
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -1432,6 +1423,7 @@ async fn handle_initialize(
         Ok(session_id) => session_id,
         Err(error) => return internal_jsonrpc_error(message, error),
     };
+    state.session_metrics.refresh();
 
     transport_response(
         StatusCode::OK,

crates/crank-community-mcp/src/app/metrics.rs

@@ -1,93 +1,107 @@
 use std::sync::Arc;
 
-use axum::http::StatusCode;
+use axum::{http::header::CONTENT_TYPE, response::Response};
+use crank_metrics::{
+    InFlightGuard, LimitStage, McpMethod, McpOutcome, McpResponseMode, record_limit_rejection,
+    record_mcp_request,
+};
 use serde_json::Value;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 
-use crate::{
-    jsonrpc::{is_notification, is_response, method_name},
-    transport::ResponseMode,
-};
+use crate::jsonrpc::{is_notification, is_response, method_name};
 
 pub(super) struct McpRequestMetrics {
-    method: &'static str,
-    response_mode: &'static str,
-    outcome: &'static str,
+    method: McpMethod,
+    response_mode: McpResponseMode,
+    outcome: McpOutcome,
 }
 
 impl McpRequestMetrics {
-    pub(super) fn new(message: &Value) -> Self {
+    pub(super) const fn invalid() -> Self {
         Self {
-            method: normalized_mcp_method(message),
-            response_mode: "unknown",
-            outcome: "rejected",
+            method: McpMethod::Invalid,
+            response_mode: McpResponseMode::Unknown,
+            outcome: McpOutcome::Aborted,
         }
     }
 
-    pub(super) fn set_response_mode(&mut self, mode: ResponseMode) -> ResponseMode {
-        self.response_mode = match mode {
-            ResponseMode::Json => "json",
-            ResponseMode::Sse => "sse",
-        };
-        mode
+    pub(super) fn new(message: &Value) -> Self {
+        Self {
+            method: normalized_mcp_method(message),
+            response_mode: McpResponseMode::Unknown,
+            outcome: McpOutcome::Aborted,
+        }
     }
 
-    pub(super) fn complete(&mut self, status: StatusCode) {
-        self.outcome = match status.as_u16() {
-            200..=299 => "success",
-            400..=499 => "client_error",
-            500..=599 => "server_error",
-            _ => "other",
+    pub(super) fn complete(&mut self, response: &Response) {
+        self.response_mode = response
+            .headers()
+            .get(CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok())
+            .map_or(McpResponseMode::Unknown, |content_type| {
+                if content_type.starts_with("application/json") {
+                    McpResponseMode::Json
+                } else if content_type.starts_with("text/event-stream") {
+                    McpResponseMode::Sse
+                } else {
+                    McpResponseMode::Unknown
+                }
+            });
+        self.outcome = if response.status().is_success() {
+            response
+                .extensions()
+                .get::<McpOutcome>()
+                .copied()
+                .unwrap_or_else(|| McpOutcome::from_http_status(response.status().as_u16()))
+        } else {
+            McpOutcome::from_http_status(response.status().as_u16())
         };
     }
+
+    #[cfg(test)]
+    pub(super) const fn outcome(&self) -> McpOutcome {
+        self.outcome
+    }
+
+    #[cfg(test)]
+    pub(super) const fn response_mode(&self) -> McpResponseMode {
+        self.response_mode
+    }
 }
 
 impl Drop for McpRequestMetrics {
     fn drop(&mut self) {
-        ::metrics::counter!(
-            "crank_mcp_requests_total",
-            "method" => self.method,
-            "response_mode" => self.response_mode,
-            "outcome" => self.outcome
-        )
-        .increment(1);
+        record_mcp_request(self.method, self.response_mode, self.outcome);
     }
 }
 
-pub(super) fn normalized_mcp_method(message: &Value) -> &'static str {
+pub(super) fn normalized_mcp_method(message: &Value) -> McpMethod {
     match method_name(message) {
-        Some("initialize") => "initialize",
-        Some("notifications/initialized") => "initialized",
-        Some("ping") => "ping",
-        Some("tools/list") => "tools_list",
-        Some("tools/call") => "tools_call",
-        Some(_) if is_notification(message) => "notification",
-        Some(_) => "unsupported",
-        None if is_response(message) => "response",
-        None => "invalid",
+        Some("initialize") => McpMethod::Initialize,
+        Some("notifications/initialized") => McpMethod::Initialized,
+        Some("ping") => McpMethod::Ping,
+        Some("tools/list") => McpMethod::ToolsList,
+        Some("tools/call") => McpMethod::ToolsCall,
+        Some(_) if is_notification(message) => McpMethod::Notification,
+        Some(_) => McpMethod::Unsupported,
+        None if is_response(message) => McpMethod::Response,
+        None => McpMethod::Invalid,
     }
 }
 
-pub(super) struct ActiveSessionGuard {
+pub(super) struct ActiveStreamGuard {
     _permit: OwnedSemaphorePermit,
+    _inflight: InFlightGuard,
 }
 
-impl ActiveSessionGuard {
+impl ActiveStreamGuard {
     pub(super) fn try_acquire(slots: &Arc<Semaphore>) -> Result<Self, ()> {
         let permit = Arc::clone(slots).try_acquire_owned().map_err(|_| {
-            ::metrics::counter!(
-                "crank_runtime_limit_rejections_total",
-                "stage" => "mcp_session"
-            )
-            .increment(1);
+            record_limit_rejection(LimitStage::McpStream);
         })?;
-        ::metrics::gauge!("crank_mcp_active_sessions").increment(1.0);
-        Ok(Self { _permit: permit })
-    }
-}
-
-impl Drop for ActiveSessionGuard {
-    fn drop(&mut self) {
-        ::metrics::gauge!("crank_mcp_active_sessions").decrement(1.0);
+        Ok(Self {
+            _permit: permit,
+            _inflight: InFlightGuard::mcp_stream(),
+        })
     }
 }

crates/crank-community-mcp/src/app/tests.rs

@@ -5,6 +5,7 @@ use std::{
 
 use axum::body::to_bytes;
 use crank_core::InvocationStatus;
+use crank_metrics::{McpMethod, McpOutcome, McpResponseMode};
 use crank_observability::{
     ObservabilityConfig, OperationalIncident, RedactionLimits, ServiceIdentity,
     operational_incident_total,
@@ -16,16 +17,20 @@ use serde_json::{Value, json};
 use tracing_subscriber::fmt::MakeWriter;
 
 use super::{
-    ResponseMode, metrics::normalized_mcp_method, observe_invocation_history_outcome,
-    tool_error_response,
+    ResponseMode,
+    metrics::{McpRequestMetrics, normalized_mcp_method},
+    observe_invocation_history_outcome, tool_error_response,
 };
-use crate::jsonrpc::CURRENT_PROTOCOL_VERSION;
+use crate::jsonrpc::{CURRENT_PROTOCOL_VERSION, jsonrpc_error};
 use crate::tool_error::generic_tool_error_contract;
+use crate::transport::transport_response;
 
 #[tokio::test]
 async fn tool_error_response_includes_structured_context() {
+    let message = json!({"jsonrpc": "2.0", "id": "req-1", "method": "tools/call"});
+    let mut request_metrics = McpRequestMetrics::new(&message);
     let response = tool_error_response(
-        &json!({"jsonrpc": "2.0", "id": "req-1"}),
+        &message,
         ResponseMode::Json,
         CURRENT_PROTOCOL_VERSION,
         generic_tool_error_contract(
@@ -36,6 +41,12 @@ async fn tool_error_response_includes_structured_context() {
             Some("Проверьте параметры вызова инструмента."),
         ),
     );
+    assert_eq!(
+        response.extensions().get::<McpOutcome>(),
+        Some(&McpOutcome::ToolError)
+    );
+    request_metrics.complete(&response);
+    assert_eq!(request_metrics.outcome(), McpOutcome::ToolError);
 
     let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
     let payload: Value = serde_json::from_slice(&body).unwrap();
@@ -53,6 +64,67 @@ async fn tool_error_response_includes_structured_context() {
     );
 }
 
+#[test]
+fn jsonrpc_error_over_http_200_is_not_counted_as_success() {
+    let message = json!({"jsonrpc": "2.0", "id": 1, "method": "unsupported"});
+    let mut request_metrics = McpRequestMetrics::new(&message);
+    let response = transport_response(
+        axum::http::StatusCode::OK,
+        jsonrpc_error(json!(1), -32601, "unsupported"),
+        ResponseMode::Json,
+        None,
+        Some(CURRENT_PROTOCOL_VERSION),
+    );
+
+    request_metrics.complete(&response);
+
+    assert_eq!(request_metrics.outcome(), McpOutcome::JsonRpcError);
+}
+
+#[test]
+fn metric_uses_actual_json_response_when_sse_request_falls_back() {
+    let message = json!({"jsonrpc": "2.0", "id": 1, "method": "unsupported"});
+    let mut request_metrics = McpRequestMetrics::new(&message);
+    let response = transport_response(
+        axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+        jsonrpc_error(json!(1), -32603, "internal error"),
+        ResponseMode::Json,
+        None,
+        Some(CURRENT_PROTOCOL_VERSION),
+    );
+
+    request_metrics.complete(&response);
+
+    assert_eq!(request_metrics.response_mode(), McpResponseMode::Json);
+    assert_eq!(request_metrics.outcome(), McpOutcome::ServerError);
+}
+
+#[test]
+fn transport_failure_takes_priority_over_jsonrpc_payload() {
+    let message = json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call"});
+    let mut request_metrics = McpRequestMetrics::new(&message);
+    let response = transport_response(
+        axum::http::StatusCode::TOO_MANY_REQUESTS,
+        jsonrpc_error(json!(1), -32000, "rate limited"),
+        ResponseMode::Json,
+        None,
+        Some(CURRENT_PROTOCOL_VERSION),
+    );
+
+    request_metrics.complete(&response);
+
+    assert_eq!(request_metrics.outcome(), McpOutcome::ClientError);
+}
+
+#[test]
+fn unfinished_request_is_classified_as_aborted() {
+    let message = json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call"});
+    let request_metrics = McpRequestMetrics::new(&message);
+
+    assert_eq!(request_metrics.response_mode(), McpResponseMode::Unknown);
+    assert_eq!(request_metrics.outcome(), McpOutcome::Aborted);
+}
+
 #[test]
 fn emits_bounded_history_loss_incident() {
     let writer = SharedLogWriter::default();
@@ -96,19 +168,19 @@ fn emits_bounded_history_loss_incident() {
 fn mcp_metric_method_is_always_from_a_closed_set() {
     assert_eq!(
         normalized_mcp_method(&json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call"})),
-        "tools_call"
+        McpMethod::ToolsCall
     );
     assert_eq!(
         normalized_mcp_method(
             &json!({"jsonrpc": "2.0", "id": 2, "method": "customer-controlled-method"})
         ),
-        "unsupported"
+        McpMethod::Unsupported
     );
     assert_eq!(
         normalized_mcp_method(
             &json!({"jsonrpc": "2.0", "method": "customer-controlled-notification"})
         ),
-        "notification"
+        McpMethod::Notification
     );
 }
 

crates/crank-community-mcp/src/catalog.rs

@@ -339,10 +339,11 @@ fn record_catalog_metrics(metrics: impl Iterator<Item = CatalogMetrics>) {
         aggregate
     });
 
-    metrics::gauge!("crank_catalog_tools").set(aggregate.tool_count as f64);
-    metrics::gauge!("crank_catalog_estimated_context_tokens")
-        .set(aggregate.estimated_context_tokens as f64);
-    metrics::gauge!("crank_catalog_warnings").set(aggregate.warning_count as f64);
+    crank_metrics::set_catalog(
+        aggregate.tool_count,
+        aggregate.estimated_context_tokens,
+        aggregate.warning_count,
+    );
 }
 
 fn now_unix_ms() -> u64 {

crates/crank-community-mcp/src/session.rs

@@ -9,9 +9,12 @@ use sqlx::{
 };
 use thiserror::Error;
 use time::OffsetDateTime;
-use tokio::sync::RwLock;
+use tokio::sync::{RwLock, mpsc};
+use tracing::{info, warn};
 use uuid::Uuid;
 
+const ACTIVE_SESSION_COUNT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct SessionState {
     pub id: String,
@@ -54,10 +57,72 @@ pub trait TransportSessionStore: Send + Sync {
     async fn delete(&self, session_id: &str) -> Result<bool, SessionStoreError>;
 
     async fn cleanup_expired(&self, now: OffsetDateTime) -> Result<u64, SessionStoreError>;
+
+    async fn active_count(&self, now: OffsetDateTime) -> Result<u64, SessionStoreError>;
 }
 
 pub type SharedSessionStore = Arc<dyn TransportSessionStore>;
 
+#[derive(Clone)]
+pub struct ActiveSessionMetrics {
+    refresh_tx: mpsc::Sender<()>,
+}
+
+impl ActiveSessionMetrics {
+    pub fn start(sessions: SharedSessionStore) -> Self {
+        let (refresh_tx, mut refresh_rx) = mpsc::channel(1);
+        tokio::spawn(async move {
+            while refresh_rx.recv().await.is_some() {
+                match tokio::time::timeout(
+                    ACTIVE_SESSION_COUNT_TIMEOUT,
+                    sessions.active_count(OffsetDateTime::now_utc()),
+                )
+                .await
+                {
+                    Ok(Ok(count)) => crank_metrics::set_mcp_active_sessions(count),
+                    Ok(Err(_)) | Err(_) => {
+                        warn!(
+                            name: "mcp.active_session_metrics.refresh_failed",
+                            error_category = "session_store",
+                            "active session metrics refresh failed"
+                        );
+                    }
+                }
+            }
+        });
+        let metrics = Self { refresh_tx };
+        metrics.refresh();
+        metrics
+    }
+
+    pub fn refresh(&self) {
+        let _ = self.refresh_tx.try_send(());
+    }
+}
+
+pub fn spawn_session_cleanup(
+    sessions: SharedSessionStore,
+    metrics: ActiveSessionMetrics,
+    cleanup_interval: std::time::Duration,
+) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(cleanup_interval);
+        loop {
+            interval.tick().await;
+            match sessions.cleanup_expired(OffsetDateTime::now_utc()).await {
+                Ok(removed) if removed > 0 => {
+                    info!(name: "mcp.session_cleanup.completed", removed);
+                }
+                Ok(_) => {}
+                Err(_) => {
+                    warn!(name: "mcp.session_cleanup.failed", error_category = "session_store");
+                }
+            }
+            metrics.refresh();
+        }
+    });
+}
+
 #[derive(Clone, Debug)]
 pub struct PostgresTransportSessionStore {
     pool: PgPool,
@@ -176,6 +241,17 @@ impl TransportSessionStore for InMemorySessionStore {
         guard.retain(|_, session| !is_expired(session, now));
         Ok(u64::try_from(before.saturating_sub(guard.len())).unwrap_or(u64::MAX))
     }
+
+    async fn active_count(&self, now: OffsetDateTime) -> Result<u64, SessionStoreError> {
+        let guard = self.inner.read().await;
+        Ok(u64::try_from(
+            guard
+                .values()
+                .filter(|session| !is_expired(session, now))
+                .count(),
+        )
+        .unwrap_or(u64::MAX))
+    }
 }
 
 #[async_trait]
@@ -319,6 +395,22 @@ impl TransportSessionStore for PostgresTransportSessionStore {
 
         Ok(result.rows_affected())
     }
+
+    async fn active_count(&self, now: OffsetDateTime) -> Result<u64, SessionStoreError> {
+        let row = query(
+            "select count(*)::bigint as active_count
+             from mcp_transport_sessions
+             where expires_at is null or expires_at > $1::timestamptz",
+        )
+        .bind(now)
+        .fetch_one(&self.pool)
+        .await
+        .map_err(|error| SessionStoreError {
+            details: error.to_string(),
+        })?;
+        let count = row.get::<i64, _>("active_count");
+        Ok(u64::try_from(count).unwrap_or_default())
+    }
 }
 
 async fn apply_postgres_migrations(pool: &PgPool) -> Result<(), SessionStoreError> {

crates/crank-community-mcp/src/transport.rs

@@ -11,6 +11,7 @@ use axum::{
         sse::{Event, KeepAlive, Sse},
     },
 };
+use crank_metrics::McpOutcome;
 use futures_util::stream;
 use reqwest::Url;
 use serde_json::Value;
@@ -223,7 +224,11 @@ pub(super) fn json_response(
     session_id: Option<&str>,
     protocol_version: Option<&str>,
 ) -> Response {
+    let outcome = payload_outcome(&payload);
     let mut response = (status, Json(payload)).into_response();
+    if let Some(outcome) = outcome {
+        response.extensions_mut().insert(outcome);
+    }
 
     if let Some(session_id) = session_id {
         response.headers_mut().insert(
@@ -251,16 +256,35 @@ pub(super) fn transport_response(
     session_id: Option<&str>,
     protocol_version: Option<&str>,
 ) -> Response {
+    let outcome = payload_outcome(&payload);
     if status == StatusCode::OK && matches!(response_mode, ResponseMode::Sse) {
         let payload = payload.to_string();
         let stream = stream::once(async move { Ok(Event::default().data(payload)) });
 
-        return sse_response(status, stream, session_id, protocol_version);
+        let mut response = sse_response(status, stream, session_id, protocol_version);
+        if let Some(outcome) = outcome {
+            response.extensions_mut().insert(outcome);
+        }
+        return response;
     }
 
     json_response(status, payload, session_id, protocol_version)
 }
 
+fn payload_outcome(payload: &Value) -> Option<McpOutcome> {
+    if payload.get("error").is_some() {
+        return Some(McpOutcome::JsonRpcError);
+    }
+    if payload
+        .pointer("/result/isError")
+        .and_then(Value::as_bool)
+        .unwrap_or(false)
+    {
+        return Some(McpOutcome::ToolError);
+    }
+    None
+}
+
 pub(super) fn with_request_id_header(mut response: Response, request_id: &str) -> Response {
     if let Ok(value) = HeaderValue::from_str(request_id) {
         response.headers_mut().insert(HEADER_X_REQUEST_ID, value);

crates/crank-community-mcp/tests/integration/session.rs

@@ -133,5 +133,9 @@ async fn postgres_transport_session_cleanup_removes_abandoned_expired_rows() {
         .unwrap();
 
     assert_eq!(store.cleanup_expired(now).await.unwrap(), 1);
+    assert_eq!(store.active_count(now).await.unwrap(), 1);
     assert!(store.get(&active).await.unwrap().is_some());
+
+    assert!(store.delete(&active).await.unwrap());
+    assert_eq!(store.active_count(now).await.unwrap(), 0);
 }

crates/crank-community-mcp/tests/session_metrics.rs

@@ -0,0 +1,76 @@
+use std::{sync::Arc, time::Duration};
+
+use crank_community_mcp::session::{
+    ActiveSessionMetrics, InMemorySessionStore, SharedSessionStore, TransportSessionStore,
+    spawn_session_cleanup,
+};
+use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter};
+use time::OffsetDateTime;
+
+#[tokio::test]
+async fn active_session_sampler_tracks_create_delete_and_expiry() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    recorder
+        .install()
+        .expect("isolated integration test recorder");
+
+    let store = Arc::new(InMemorySessionStore::default());
+    let sessions: SharedSessionStore = store.clone();
+    let sampler = ActiveSessionMetrics::start(Arc::clone(&sessions));
+    assert_gauge(&sampler, &snapshotter, 0.0).await;
+
+    let now = OffsetDateTime::now_utc();
+    store
+        .create(
+            "2025-11-25",
+            "default",
+            "agent",
+            false,
+            now,
+            Some(now + time::Duration::milliseconds(100)),
+        )
+        .await
+        .unwrap();
+    assert_gauge(&sampler, &snapshotter, 1.0).await;
+
+    spawn_session_cleanup(sessions, sampler, Duration::from_millis(10));
+    assert_gauge_without_refresh(&snapshotter, 0.0).await;
+}
+
+async fn assert_gauge_without_refresh(snapshotter: &Snapshotter, expected: f64) {
+    for _ in 0..100 {
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        if gauge_value(snapshotter) == Some(expected) {
+            return;
+        }
+    }
+    panic!("active session gauge did not become {expected} through cleanup");
+}
+
+async fn assert_gauge(sampler: &ActiveSessionMetrics, snapshotter: &Snapshotter, expected: f64) {
+    for _ in 0..50 {
+        sampler.refresh();
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        if gauge_value(snapshotter) == Some(expected) {
+            return;
+        }
+    }
+    panic!("active session gauge did not become {expected}");
+}
+
+fn gauge_value(snapshotter: &Snapshotter) -> Option<f64> {
+    snapshotter
+        .snapshot()
+        .into_vec()
+        .into_iter()
+        .find_map(|(key, _, _, value)| {
+            if key.key().name() != "crank_mcp_active_sessions" {
+                return None;
+            }
+            match value {
+                DebugValue::Gauge(value) => Some(value.into_inner()),
+                _ => None,
+            }
+        })
+}

crates/crank-community-mcp/tests/unit/session.rs

@@ -93,8 +93,12 @@ async fn cleanup_removes_only_expired_sessions() {
         .unwrap();
 
     assert_eq!(store.cleanup_expired(now).await.unwrap(), 1);
+    assert_eq!(store.active_count(now).await.unwrap(), 1);
     assert!(store.get(&expired).await.unwrap().is_none());
     assert!(store.get(&active).await.unwrap().is_some());
+
+    assert!(store.delete(&active).await.unwrap());
+    assert_eq!(store.active_count(now).await.unwrap(), 0);
 }
 
 #[test]

crates/crank-metrics/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "crank-metrics"
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+publish.workspace = true
+version.workspace = true
+
+[lib]
+path = "src/lib.rs"
+
+[dependencies]
+metrics.workspace = true
+
+[dev-dependencies]
+metrics-util = "0.20.4"

crates/crank-metrics/src/labels.rs

@@ -0,0 +1,431 @@
+macro_rules! string_enum {
+    (
+        $(#[$meta:meta])*
+        pub enum $name:ident {
+            $($variant:ident => $value:literal),+ $(,)?
+        }
+    ) => {
+        $(#[$meta])*
+        #[derive(Clone, Copy, Debug, Eq, PartialEq)]
+        pub enum $name {
+            $($variant),+
+        }
+
+        impl $name {
+            pub const fn as_str(self) -> &'static str {
+                match self {
+                    $(Self::$variant => $value),+
+                }
+            }
+        }
+    };
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct HttpRoute(&'static str);
+
+impl HttpRoute {
+    pub const fn unmatched() -> Self {
+        Self("unmatched")
+    }
+
+    pub fn from_matched_path(path: &str) -> Self {
+        match path {
+            "/health" => Self("/health"),
+            "/ready" => Self("/ready"),
+            "/api/auth/login" => Self("/api/auth/login"),
+            "/api/auth/logout" => Self("/api/auth/logout"),
+            "/api/auth/session" => Self("/api/auth/session"),
+            "/api/auth/profile" => Self("/api/auth/profile"),
+            "/api/auth/password" => Self("/api/auth/password"),
+            "/api/admin/capabilities" => Self("/api/admin/capabilities"),
+            "/api/admin/workspaces" => Self("/api/admin/workspaces"),
+            "/api/admin/workspaces/{workspace_id}" => Self("/api/admin/workspaces/{workspace_id}"),
+            "/api/admin/workspaces/{workspace_id}/operations" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations")
+            }
+            "/api/admin/workspaces/{workspace_id}/imports/openapi/preview" => {
+                Self("/api/admin/workspaces/{workspace_id}/imports/openapi/preview")
+            }
+            "/api/admin/workspaces/{workspace_id}/imports/openapi/{job_id}/create" => {
+                Self("/api/admin/workspaces/{workspace_id}/imports/openapi/{job_id}/create")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/analyze-quality" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/analyze-quality")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/import" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/import")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions/{version}" => {
+                Self(
+                    "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions/{version}",
+                )
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/publish" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/publish")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/archive" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/archive")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/test-runs" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/test-runs")
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/input-json" => {
+                Self(
+                    "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/input-json",
+                )
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/output-json" => {
+                Self(
+                    "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/output-json",
+                )
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/drafts/generate" => {
+                Self(
+                    "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/drafts/generate",
+                )
+            }
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/export" => {
+                Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/export")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/tool-search/preview" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/tool-search/preview")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/versions/{version}" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/versions/{version}")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/bindings" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/bindings")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/publish" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/publish")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/unpublish" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/unpublish")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/archive" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/archive")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys" => {
+                Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys")
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}/revoke" => {
+                Self(
+                    "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}/revoke",
+                )
+            }
+            "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}" => {
+                Self(
+                    "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}",
+                )
+            }
+            "/api/admin/workspaces/{workspace_id}/auth-profiles" => {
+                Self("/api/admin/workspaces/{workspace_id}/auth-profiles")
+            }
+            "/api/admin/workspaces/{workspace_id}/auth-profiles/{auth_profile_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/auth-profiles/{auth_profile_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/upstreams" => {
+                Self("/api/admin/workspaces/{workspace_id}/upstreams")
+            }
+            "/api/admin/workspaces/{workspace_id}/upstreams/{upstream_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/upstreams/{upstream_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/secrets" => {
+                Self("/api/admin/workspaces/{workspace_id}/secrets")
+            }
+            "/api/admin/workspaces/{workspace_id}/secrets/{secret_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/secrets/{secret_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/secrets/{secret_id}/rotate" => {
+                Self("/api/admin/workspaces/{workspace_id}/secrets/{secret_id}/rotate")
+            }
+            "/api/admin/workspaces/{workspace_id}/export" => {
+                Self("/api/admin/workspaces/{workspace_id}/export")
+            }
+            "/api/admin/workspaces/{workspace_id}/logs" => {
+                Self("/api/admin/workspaces/{workspace_id}/logs")
+            }
+            "/api/admin/workspaces/{workspace_id}/logs/{log_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/logs/{log_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/approvals" => {
+                Self("/api/admin/workspaces/{workspace_id}/approvals")
+            }
+            "/api/admin/workspaces/{workspace_id}/approvals/{approval_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/approvals/{approval_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/usage" => {
+                Self("/api/admin/workspaces/{workspace_id}/usage")
+            }
+            "/api/admin/workspaces/{workspace_id}/usage/operations/{operation_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/usage/operations/{operation_id}")
+            }
+            "/api/admin/workspaces/{workspace_id}/usage/agents/{agent_id}" => {
+                Self("/api/admin/workspaces/{workspace_id}/usage/agents/{agent_id}")
+            }
+            "/v1/{workspace_slug}/{agent_slug}" => Self("/v1/{workspace_slug}/{agent_slug}"),
+            "/v1/{workspace_slug}/{agent_slug}/approvals" => {
+                Self("/v1/{workspace_slug}/{agent_slug}/approvals")
+            }
+            "/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/approve" => {
+                Self("/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/approve")
+            }
+            "/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}" => {
+                Self("/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}")
+            }
+            "/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/deny" => {
+                Self("/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/deny")
+            }
+            _ => Self::unmatched(),
+        }
+    }
+
+    pub const fn as_str(self) -> &'static str {
+        self.0
+    }
+}
+
+string_enum! {
+    pub enum HttpMethod {
+        Get => "GET",
+        Post => "POST",
+        Put => "PUT",
+        Patch => "PATCH",
+        Delete => "DELETE",
+        Options => "OPTIONS",
+        Head => "HEAD",
+        Connect => "CONNECT",
+        Trace => "TRACE",
+        Other => "OTHER",
+    }
+}
+
+impl HttpMethod {
+    pub fn classify(method: &str) -> Self {
+        match method {
+            "GET" => Self::Get,
+            "POST" => Self::Post,
+            "PUT" => Self::Put,
+            "PATCH" => Self::Patch,
+            "DELETE" => Self::Delete,
+            "OPTIONS" => Self::Options,
+            "HEAD" => Self::Head,
+            "CONNECT" => Self::Connect,
+            "TRACE" => Self::Trace,
+            _ => Self::Other,
+        }
+    }
+}
+
+impl McpOutcome {
+    pub const fn from_http_status(status: u16) -> Self {
+        match status {
+            200..=299 => Self::Success,
+            400..=499 => Self::ClientError,
+            500..=599 => Self::ServerError,
+            _ => Self::Other,
+        }
+    }
+}
+
+string_enum! {
+    pub enum HttpStatusClass {
+        Informational => "1xx",
+        Success => "2xx",
+        Redirection => "3xx",
+        ClientError => "4xx",
+        ServerError => "5xx",
+        Other => "other",
+    }
+}
+
+impl HttpStatusClass {
+    pub const fn from_status(status: u16) -> Self {
+        match status {
+            100..=199 => Self::Informational,
+            200..=299 => Self::Success,
+            300..=399 => Self::Redirection,
+            400..=499 => Self::ClientError,
+            500..=599 => Self::ServerError,
+            _ => Self::Other,
+        }
+    }
+}
+
+string_enum! {
+    pub enum McpMethod {
+        Initialize => "initialize",
+        Initialized => "initialized",
+        Ping => "ping",
+        ToolsList => "tools_list",
+        ToolsCall => "tools_call",
+        Notification => "notification",
+        Unsupported => "unsupported",
+        Response => "response",
+        Invalid => "invalid",
+    }
+}
+
+string_enum! {
+    pub enum McpResponseMode {
+        Json => "json",
+        Sse => "sse",
+        Unknown => "unknown",
+    }
+}
+
+string_enum! {
+    pub enum McpOutcome {
+        Success => "success",
+        ClientError => "client_error",
+        ServerError => "server_error",
+        JsonRpcError => "jsonrpc_error",
+        ToolError => "tool_error",
+        Aborted => "aborted",
+        Other => "other",
+    }
+}
+
+string_enum! {
+    pub enum InvocationSource {
+        Internal => "internal",
+        AdminTestRun => "admin_test_run",
+        AgentToolCall => "agent_tool_call",
+    }
+}
+
+string_enum! {
+    pub enum ToolOutcome {
+        Success => "success",
+        Error => "error",
+        Aborted => "aborted",
+    }
+}
+
+string_enum! {
+    pub enum ToolErrorKind {
+        None => "none",
+        Schema => "schema",
+        Mapping => "mapping",
+        RestAdapter => "rest_adapter",
+        ProtocolAdapter => "protocol_adapter",
+        UnsupportedProtocol => "unsupported_protocol",
+        UnsupportedExecutionMode => "unsupported_execution_mode",
+        ConcurrencyLimit => "concurrency_limit",
+        InvalidPreparedRequest => "invalid_prepared_request",
+        ConfirmationRequired => "confirmation_required",
+        InvalidConfirmationToken => "invalid_confirmation_token",
+        ConfirmationStore => "confirmation_store",
+        IdempotencyStore => "idempotency_store",
+        IdempotencyInProgress => "idempotency_in_progress",
+        IdempotencyConflict => "idempotency_conflict",
+        IdempotencyOutcomeUnknown => "idempotency_outcome_unknown",
+        MissingAuthProfile => "missing_auth_profile",
+        MissingSecret => "missing_secret",
+        MissingSecretVersion => "missing_secret_version",
+        InvalidAuthSecret => "invalid_auth_secret",
+        SecretCrypto => "secret_crypto",
+        Aborted => "aborted",
+    }
+}
+
+string_enum! {
+    pub enum UpstreamOperationKind {
+        Rest => "rest",
+    }
+}
+
+string_enum! {
+    pub enum UpstreamOutcome {
+        Success => "success",
+        ClientError => "client_error",
+        ServerError => "server_error",
+        UnexpectedStatus => "unexpected_status",
+        Timeout => "timeout",
+        TransportError => "transport_error",
+        ResponseTooLarge => "response_too_large",
+        Rejected => "rejected",
+        WindowExpired => "window_expired",
+        InvalidResponse => "invalid_response",
+        InvalidRequest => "invalid_request",
+        Configuration => "configuration",
+        Aborted => "aborted",
+    }
+}
+
+string_enum! {
+    pub enum LimitStage {
+        Concurrency => "concurrency",
+        RateLimit => "rate_limit",
+        McpStream => "mcp_stream",
+    }
+}
+
+string_enum! {
+    pub enum CacheOutcome {
+        Hit => "hit",
+        Miss => "miss",
+        ReadError => "read_error",
+        DecodeError => "decode_error",
+        EvictError => "evict_error",
+        Stored => "stored",
+        WriteError => "write_error",
+    }
+}
+
+string_enum! {
+    pub enum IdempotencyOutcome {
+        Execute => "execute",
+        Replay => "replay",
+        Completed => "completed",
+        Conflict => "conflict",
+        InProgress => "in_progress",
+        OutcomeUnknown => "outcome_unknown",
+        StoreUnavailable => "store_unavailable",
+        Error => "error",
+    }
+}
+
+string_enum! {
+    pub enum ConfirmationOutcome {
+        Approved => "approved",
+        Required => "required",
+        InvalidToken => "invalid_token",
+        StoreUnavailable => "store_unavailable",
+        Error => "error",
+    }
+}
+
+string_enum! {
+    pub enum DbPoolState {
+        Idle => "idle",
+        Used => "used",
+    }
+}
+
+string_enum! {
+    pub enum SignalType {
+        Trace => "trace",
+        InvocationHistory => "invocation_history",
+    }
+}
+
+string_enum! {
+    pub enum Exporter {
+        Otlp => "otlp",
+        Postgres => "postgres",
+    }
+}

crates/crank-metrics/src/lib.rs

@@ -0,0 +1,26 @@
+//! Закрытый семантический контракт метрик Crank.
+//!
+//! Product crate передают только типизированные значения. Имена метрик,
+//! ключи и значения labels сосредоточены здесь, поэтому пользовательские
+//! идентификаторы и тексты нельзя случайно превратить во временные ряды.
+
+mod labels;
+mod record;
+mod schema;
+
+pub use labels::{
+    CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute,
+    HttpStatusClass, IdempotencyOutcome, InvocationSource, LimitStage, McpMethod, McpOutcome,
+    McpResponseMode, SignalType, ToolErrorKind, ToolOutcome, UpstreamOperationKind,
+    UpstreamOutcome,
+};
+pub use record::{
+    InFlightGuard, ToolInvocationMetrics, UpstreamRequestMetrics, initialize_gauges,
+    record_cache_outcome, record_confirmation_outcome, record_export_failure, record_http_request,
+    record_idempotency_outcome, record_invocation_history_lost, record_limit_rejection,
+    record_mcp_request, record_tool_invocation, record_upstream_request, set_catalog,
+    set_db_pool_connections, set_mcp_active_sessions,
+};
+pub use schema::{
+    DURATION_BUCKETS_SECONDS, MetricDefinition, MetricKind, MetricUnit, metric_schema,
+};

crates/crank-metrics/src/record.rs

@@ -0,0 +1,243 @@
+use std::time::{Duration, Instant};
+
+use metrics::Gauge;
+
+use crate::{
+    CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute,
+    HttpStatusClass, IdempotencyOutcome, InvocationSource, LimitStage, McpMethod, McpOutcome,
+    McpResponseMode, SignalType, ToolErrorKind, ToolOutcome, UpstreamOperationKind,
+    UpstreamOutcome,
+};
+
+pub fn record_http_request(
+    route: HttpRoute,
+    method: HttpMethod,
+    status: HttpStatusClass,
+    duration: Duration,
+) {
+    metrics::counter!(
+        "crank_http_requests_total",
+        "route" => route.as_str(),
+        "method" => method.as_str(),
+        "status_class" => status.as_str()
+    )
+    .increment(1);
+    metrics::histogram!(
+        "crank_http_request_duration_seconds",
+        "route" => route.as_str(),
+        "method" => method.as_str()
+    )
+    .record(duration.as_secs_f64());
+}
+
+pub fn record_mcp_request(method: McpMethod, response_mode: McpResponseMode, outcome: McpOutcome) {
+    metrics::counter!(
+        "crank_mcp_requests_total",
+        "method" => method.as_str(),
+        "response_mode" => response_mode.as_str(),
+        "outcome" => outcome.as_str()
+    )
+    .increment(1);
+}
+
+pub fn set_mcp_active_sessions(count: u64) {
+    metrics::gauge!("crank_mcp_active_sessions").set(count as f64);
+}
+
+pub fn record_tool_invocation(
+    source: InvocationSource,
+    outcome: ToolOutcome,
+    error_kind: ToolErrorKind,
+    duration: Duration,
+) {
+    metrics::counter!(
+        "crank_tool_invocations_total",
+        "source" => source.as_str(),
+        "outcome" => outcome.as_str(),
+        "error_kind" => error_kind.as_str()
+    )
+    .increment(1);
+    metrics::histogram!(
+        "crank_tool_invocation_duration_seconds",
+        "source" => source.as_str(),
+        "outcome" => outcome.as_str()
+    )
+    .record(duration.as_secs_f64());
+}
+
+pub fn record_upstream_request(
+    operation_kind: UpstreamOperationKind,
+    outcome: UpstreamOutcome,
+    duration: Duration,
+) {
+    metrics::counter!(
+        "crank_upstream_requests_total",
+        "operation_kind" => operation_kind.as_str(),
+        "outcome" => outcome.as_str()
+    )
+    .increment(1);
+    metrics::histogram!(
+        "crank_upstream_request_duration_seconds",
+        "operation_kind" => operation_kind.as_str(),
+        "outcome" => outcome.as_str()
+    )
+    .record(duration.as_secs_f64());
+}
+
+pub struct ToolInvocationMetrics {
+    source: InvocationSource,
+    started_at: Option<Instant>,
+}
+
+impl ToolInvocationMetrics {
+    pub fn start(source: InvocationSource) -> Self {
+        Self {
+            source,
+            started_at: Some(Instant::now()),
+        }
+    }
+
+    pub fn complete(mut self, outcome: ToolOutcome, error_kind: ToolErrorKind) {
+        self.finish(outcome, error_kind);
+    }
+
+    fn finish(&mut self, outcome: ToolOutcome, error_kind: ToolErrorKind) {
+        if let Some(started_at) = self.started_at.take() {
+            record_tool_invocation(self.source, outcome, error_kind, started_at.elapsed());
+        }
+    }
+}
+
+impl Drop for ToolInvocationMetrics {
+    fn drop(&mut self) {
+        self.finish(ToolOutcome::Aborted, ToolErrorKind::Aborted);
+    }
+}
+
+pub struct UpstreamRequestMetrics {
+    operation_kind: UpstreamOperationKind,
+    started_at: Option<Instant>,
+}
+
+impl UpstreamRequestMetrics {
+    pub fn start(operation_kind: UpstreamOperationKind) -> Self {
+        Self {
+            operation_kind,
+            started_at: Some(Instant::now()),
+        }
+    }
+
+    pub fn complete(mut self, outcome: UpstreamOutcome) {
+        self.finish(outcome);
+    }
+
+    fn finish(&mut self, outcome: UpstreamOutcome) {
+        if let Some(started_at) = self.started_at.take() {
+            record_upstream_request(self.operation_kind, outcome, started_at.elapsed());
+        }
+    }
+}
+
+impl Drop for UpstreamRequestMetrics {
+    fn drop(&mut self) {
+        self.finish(UpstreamOutcome::Aborted);
+    }
+}
+
+pub fn record_limit_rejection(stage: LimitStage) {
+    metrics::counter!(
+        "crank_runtime_limit_rejections_total",
+        "stage" => stage.as_str()
+    )
+    .increment(1);
+}
+
+pub fn record_cache_outcome(outcome: CacheOutcome) {
+    metrics::counter!(
+        "crank_runtime_cache_total",
+        "outcome" => outcome.as_str()
+    )
+    .increment(1);
+}
+
+pub fn record_idempotency_outcome(outcome: IdempotencyOutcome) {
+    metrics::counter!(
+        "crank_idempotency_total",
+        "outcome" => outcome.as_str()
+    )
+    .increment(1);
+}
+
+pub fn record_confirmation_outcome(outcome: ConfirmationOutcome) {
+    metrics::counter!(
+        "crank_confirmation_total",
+        "outcome" => outcome.as_str()
+    )
+    .increment(1);
+}
+
+pub fn set_db_pool_connections(state: DbPoolState, count: u32) {
+    metrics::gauge!(
+        "crank_db_pool_connections",
+        "state" => state.as_str()
+    )
+    .set(f64::from(count));
+}
+
+pub fn set_catalog(tool_count: usize, estimated_context_tokens: usize, warnings: usize) {
+    metrics::gauge!("crank_catalog_tools").set(tool_count as f64);
+    metrics::gauge!("crank_catalog_estimated_context_tokens").set(estimated_context_tokens as f64);
+    metrics::gauge!("crank_catalog_warnings").set(warnings as f64);
+}
+
+pub fn record_invocation_history_lost() {
+    metrics::counter!("crank_invocation_history_lost_total").increment(1);
+}
+
+pub fn record_export_failure(signal: SignalType, exporter: Exporter) {
+    metrics::counter!(
+        "crank_telemetry_export_failures_total",
+        "signal_type" => signal.as_str(),
+        "exporter" => exporter.as_str()
+    )
+    .increment(1);
+}
+
+pub fn initialize_gauges() {
+    metrics::gauge!("crank_http_inflight").set(0.0);
+    metrics::gauge!("crank_mcp_active_sessions").set(0.0);
+    metrics::gauge!("crank_mcp_active_streams").set(0.0);
+    metrics::gauge!("crank_runtime_inflight").set(0.0);
+    set_db_pool_connections(DbPoolState::Idle, 0);
+    set_db_pool_connections(DbPoolState::Used, 0);
+    set_catalog(0, 0, 0);
+}
+
+pub struct InFlightGuard {
+    gauge: Gauge,
+}
+
+impl InFlightGuard {
+    pub fn http() -> Self {
+        Self::increment(metrics::gauge!("crank_http_inflight"))
+    }
+
+    pub fn runtime() -> Self {
+        Self::increment(metrics::gauge!("crank_runtime_inflight"))
+    }
+
+    pub fn mcp_stream() -> Self {
+        Self::increment(metrics::gauge!("crank_mcp_active_streams"))
+    }
+
+    fn increment(gauge: Gauge) -> Self {
+        gauge.increment(1.0);
+        Self { gauge }
+    }
+}
+
+impl Drop for InFlightGuard {
+    fn drop(&mut self) {
+        self.gauge.decrement(1.0);
+    }
+}

crates/crank-metrics/src/schema.rs

@@ -48,7 +48,12 @@ const METRIC_SCHEMA: &[MetricDefinition] = &[
     gauge(
         "crank_mcp_active_sessions",
         &[],
-        "Active MCP transport sessions.",
+        "Active persisted MCP transport sessions.",
+    ),
+    gauge(
+        "crank_mcp_active_streams",
+        &[],
+        "MCP event streams currently being served.",
     ),
     counter(
         "crank_tool_invocations_total",
@@ -80,6 +85,21 @@ const METRIC_SCHEMA: &[MetricDefinition] = &[
         &["stage"],
         "Runtime executions rejected by a bounded limit.",
     ),
+    counter(
+        "crank_runtime_cache_total",
+        &["outcome"],
+        "Runtime response cache outcomes.",
+    ),
+    counter(
+        "crank_idempotency_total",
+        &["outcome"],
+        "Runtime idempotency outcomes.",
+    ),
+    counter(
+        "crank_confirmation_total",
+        &["outcome"],
+        "Runtime confirmation outcomes.",
+    ),
     gauge(
         "crank_db_pool_connections",
         &["state"],

crates/crank-metrics/tests/contract.rs

@@ -0,0 +1,235 @@
+use std::{collections::BTreeSet, time::Duration};
+
+use crank_metrics::{
+    CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute,
+    HttpStatusClass, IdempotencyOutcome, InFlightGuard, InvocationSource, LimitStage, McpMethod,
+    McpOutcome, McpResponseMode, SignalType, ToolErrorKind, ToolInvocationMetrics, ToolOutcome,
+    UpstreamOperationKind, UpstreamOutcome, UpstreamRequestMetrics, initialize_gauges,
+    metric_schema, record_cache_outcome, record_confirmation_outcome, record_export_failure,
+    record_http_request, record_idempotency_outcome, record_invocation_history_lost,
+    record_limit_rejection, record_mcp_request, record_tool_invocation, record_upstream_request,
+    set_catalog, set_db_pool_connections, set_mcp_active_sessions,
+};
+use metrics_util::debugging::DebuggingRecorder;
+
+#[test]
+fn schema_names_are_unique_and_labels_are_closed() {
+    let schema = metric_schema();
+    let names = schema
+        .iter()
+        .map(|metric| metric.name)
+        .collect::<BTreeSet<_>>();
+
+    assert_eq!(names.len(), schema.len());
+    assert!(names.contains("crank_runtime_cache_total"));
+    assert!(names.contains("crank_idempotency_total"));
+    assert!(names.contains("crank_confirmation_total"));
+    assert!(names.contains("crank_mcp_active_streams"));
+
+    let allowed_labels = [
+        "route",
+        "method",
+        "status_class",
+        "response_mode",
+        "outcome",
+        "source",
+        "error_kind",
+        "operation_kind",
+        "stage",
+        "state",
+        "signal_type",
+        "exporter",
+    ];
+    for metric in schema {
+        assert!(
+            metric
+                .labels
+                .iter()
+                .all(|label| allowed_labels.contains(label)),
+            "{} contains an unapproved label",
+            metric.name
+        );
+    }
+}
+
+#[test]
+fn untrusted_http_values_collapse_to_closed_variants() {
+    assert_eq!(
+        HttpRoute::from_matched_path("/api/admin/workspaces/{workspace_id}/operations").as_str(),
+        "/api/admin/workspaces/{workspace_id}/operations"
+    );
+    assert_eq!(
+        HttpRoute::from_matched_path("/customer-controlled"),
+        HttpRoute::unmatched()
+    );
+    assert_eq!(HttpMethod::classify("CUSTOM"), HttpMethod::Other);
+    assert_eq!(HttpStatusClass::from_status(999), HttpStatusClass::Other);
+}
+
+#[test]
+fn application_outcomes_are_distinct_from_transport_success() {
+    assert_eq!(McpOutcome::from_http_status(200), McpOutcome::Success);
+    assert_ne!(McpOutcome::JsonRpcError, McpOutcome::Success);
+    assert_ne!(McpOutcome::ToolError, McpOutcome::Success);
+
+    assert_eq!(CacheOutcome::Hit.as_str(), "hit");
+    assert_eq!(CacheOutcome::Miss.as_str(), "miss");
+    assert_eq!(IdempotencyOutcome::Replay.as_str(), "replay");
+    assert_eq!(IdempotencyOutcome::Conflict.as_str(), "conflict");
+    assert_eq!(ConfirmationOutcome::Approved.as_str(), "approved");
+    assert_eq!(ConfirmationOutcome::Required.as_str(), "required");
+}
+
+#[test]
+fn dropped_product_guards_record_aborted_outcomes() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+
+    metrics::with_local_recorder(&recorder, || {
+        drop(ToolInvocationMetrics::start(
+            InvocationSource::AgentToolCall,
+        ));
+        drop(UpstreamRequestMetrics::start(UpstreamOperationKind::Rest));
+    });
+
+    let snapshot = snapshotter.snapshot().into_vec();
+    assert!(snapshot.iter().any(|(key, _, _, _)| {
+        key.key().name() == "crank_tool_invocations_total"
+            && has_label(key.key(), "outcome", "aborted")
+            && has_label(key.key(), "error_kind", "aborted")
+    }));
+    assert!(snapshot.iter().any(|(key, _, _, _)| {
+        key.key().name() == "crank_upstream_requests_total"
+            && has_label(key.key(), "outcome", "aborted")
+    }));
+}
+
+fn has_label(key: &metrics::Key, name: &str, value: &str) -> bool {
+    key.labels()
+        .any(|label| label.key() == name && label.value() == value)
+}
+
+#[test]
+fn diverse_calls_cannot_create_unbounded_series() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+
+    metrics::with_local_recorder(&recorder, || {
+        for status in 200..300 {
+            let route = format!("/customer-controlled/{status}");
+            let method = format!("CUSTOM-{status}");
+            record_http_request(
+                HttpRoute::from_matched_path(&route),
+                HttpMethod::classify(&method),
+                HttpStatusClass::from_status(status),
+                Duration::from_millis(u64::from(status)),
+            );
+            record_mcp_request(
+                McpMethod::ToolsCall,
+                McpResponseMode::Json,
+                McpOutcome::ToolError,
+            );
+            record_tool_invocation(
+                InvocationSource::AgentToolCall,
+                ToolOutcome::Error,
+                ToolErrorKind::Mapping,
+                Duration::from_millis(u64::from(status)),
+            );
+            record_cache_outcome(CacheOutcome::Miss);
+            record_idempotency_outcome(IdempotencyOutcome::Replay);
+            record_confirmation_outcome(ConfirmationOutcome::Required);
+        }
+    });
+
+    let snapshot = snapshotter.snapshot().into_vec();
+    let series = snapshot
+        .iter()
+        .map(|(key, _, _, _)| format!("{:?}", key.key()))
+        .collect::<BTreeSet<_>>();
+
+    assert_eq!(series.len(), 8);
+    assert!(
+        series
+            .iter()
+            .all(|series| !series.contains("customer-controlled"))
+    );
+}
+
+#[test]
+fn every_recording_api_matches_the_declared_schema() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+
+    metrics::with_local_recorder(&recorder, || {
+        initialize_gauges();
+        record_http_request(
+            HttpRoute::from_matched_path("/api/auth/session"),
+            HttpMethod::Get,
+            HttpStatusClass::Success,
+            Duration::from_millis(1),
+        );
+        record_mcp_request(
+            McpMethod::ToolsCall,
+            McpResponseMode::Json,
+            McpOutcome::Success,
+        );
+        set_mcp_active_sessions(1);
+        let _stream = InFlightGuard::mcp_stream();
+        let _runtime = InFlightGuard::runtime();
+        record_tool_invocation(
+            InvocationSource::AgentToolCall,
+            ToolOutcome::Error,
+            ToolErrorKind::Mapping,
+            Duration::from_millis(1),
+        );
+        record_upstream_request(
+            UpstreamOperationKind::Rest,
+            UpstreamOutcome::Timeout,
+            Duration::from_millis(1),
+        );
+        record_limit_rejection(LimitStage::Concurrency);
+        record_cache_outcome(CacheOutcome::Hit);
+        record_idempotency_outcome(IdempotencyOutcome::Replay);
+        record_confirmation_outcome(ConfirmationOutcome::Required);
+        set_db_pool_connections(DbPoolState::Idle, 1);
+        set_catalog(1, 2, 3);
+        record_invocation_history_lost();
+        record_export_failure(SignalType::Trace, Exporter::Otlp);
+    });
+
+    let snapshot = snapshotter.snapshot().into_vec();
+    let actual_names = snapshot
+        .iter()
+        .map(|(key, _, _, _)| key.key().name().to_owned())
+        .collect::<BTreeSet<_>>();
+    let expected_names = metric_schema()
+        .iter()
+        .map(|definition| definition.name.to_owned())
+        .collect::<BTreeSet<_>>();
+
+    assert_eq!(actual_names, expected_names);
+    for definition in metric_schema() {
+        let actual_label_sets = snapshot
+            .iter()
+            .filter(|(key, _, _, _)| key.key().name() == definition.name)
+            .map(|(key, _, _, _)| {
+                key.key()
+                    .labels()
+                    .map(|label| label.key().to_owned())
+                    .collect::<BTreeSet<_>>()
+            })
+            .collect::<BTreeSet<_>>();
+        let expected_labels = definition
+            .labels
+            .iter()
+            .map(|label| (*label).to_owned())
+            .collect::<BTreeSet<_>>();
+
+        assert_eq!(
+            actual_label_sets,
+            BTreeSet::from([expected_labels]),
+            "recording API for {} diverges from the schema",
+            definition.name
+        );
+    }
+}

crates/crank-observability/Cargo.toml

@@ -8,6 +8,7 @@ version.workspace = true
 
 [dependencies]
 axum.workspace = true
+crank-metrics = { path = "../crank-metrics" }
 metrics.workspace = true
 metrics-exporter-prometheus.workspace = true
 opentelemetry.workspace = true
@@ -21,7 +22,7 @@ sha2.workspace = true
 subtle.workspace = true
 thiserror.workspace = true
 time.workspace = true
-tokio = { workspace = true, features = ["net"] }
+tokio = { workspace = true, features = ["io-util", "net"] }
 tracing.workspace = true
 tracing-opentelemetry.workspace = true
 tracing-subscriber.workspace = true

crates/crank-observability/src/error_reporting.rs

@@ -1,4 +1,10 @@
-use std::{borrow::Cow, collections::BTreeMap, env, fmt, future::Future, time::Duration};
+use std::{
+    borrow::Cow,
+    collections::BTreeMap,
+    env, fmt,
+    future::Future,
+    time::{Duration, SystemTime},
+};
 
 use sentry::{
     ClientInitGuard, ClientOptions,
@@ -14,6 +20,8 @@ use crate::{
 const SENTRY_DSN_ENV: &str = "CRANK_SENTRY_DSN";
 const CRITICAL_ERROR_MESSAGE: &str = "critical error";
 const SENTRY_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2);
+// Sentry serializes SystemTime as a finite f64; this keeps conservative fixed headroom.
+const MAX_SERIALIZED_TIMESTAMP_BYTES: usize = 32;
 
 tokio::task_local! {
     static REQUEST_ID: String;
@@ -63,6 +71,8 @@ pub enum SentryConfigError {
     InvalidDsn,
     #[error("CRANK_SENTRY_DSN is not valid UTF-8")]
     InvalidEnvironmentEncoding,
+    #[error("critical error event budget cannot hold the required fields")]
+    EventBudgetTooSmall,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -74,6 +84,13 @@ pub enum CriticalErrorCategory {
 }
 
 impl CriticalErrorCategory {
+    const ALL: [Self; 4] = [
+        Self::Panic,
+        Self::Startup,
+        Self::Internal,
+        Self::DataIntegrity,
+    ];
+
     pub const fn as_str(self) -> &'static str {
         match self {
             Self::Panic => "panic",
@@ -117,11 +134,14 @@ pub(crate) fn init_sentry(
     identity: &ServiceIdentity,
     limits: RedactionLimits,
     config: SentryConfig,
-) -> Option<ClientInitGuard> {
-    let dsn = config.dsn?;
+) -> Result<Option<ClientInitGuard>, SentryConfigError> {
+    let Some(dsn) = config.dsn else {
+        return Ok(None);
+    };
+    validate_critical_event_budget(identity, limits)?;
     let identity = identity.clone();
     let options = client_options(identity, limits);
-    Some(sentry::init((dsn, options)))
+    Ok(Some(sentry::init((dsn, options))))
 }
 
 fn client_options(identity: ServiceIdentity, limits: RedactionLimits) -> ClientOptions {
@@ -181,7 +201,10 @@ fn sanitize_event(
         Event {
             event_id: event.event_id,
             level: Level::Error,
-            fingerprint: Cow::Owned(vec![Cow::Borrowed(category.as_str())]),
+            fingerprint: Cow::Owned(vec![
+                Cow::Owned(identity.service().to_owned()),
+                Cow::Borrowed(category.as_str()),
+            ]),
             message: Some(CRITICAL_ERROR_MESSAGE.to_owned()),
             timestamp: event.timestamp,
             server_name: Some(Cow::Owned(identity.service().to_owned())),
@@ -212,9 +235,44 @@ fn enforce_event_budget(mut event: Event<'static>, max_event_bytes: usize) -> Ev
 
     event.tags.remove("request_id");
     event.tags.remove("trace_id");
+    debug_assert!(serialized_event_len(&event) <= max_event_bytes);
     event
 }
 
+fn validate_critical_event_budget(
+    identity: &ServiceIdentity,
+    limits: RedactionLimits,
+) -> Result<(), SentryConfigError> {
+    if required_critical_event_budget(identity, limits) > limits.max_event_bytes {
+        return Err(SentryConfigError::EventBudgetTooSmall);
+    }
+    Ok(())
+}
+
+fn required_critical_event_budget(identity: &ServiceIdentity, limits: RedactionLimits) -> usize {
+    let unbounded_limits = RedactionLimits {
+        max_event_bytes: usize::MAX,
+        ..limits
+    };
+    CriticalErrorCategory::ALL
+        .into_iter()
+        .map(|category| {
+            let event = sanitize_event(
+                Event {
+                    tags: BTreeMap::from([("category".to_owned(), category.as_str().to_owned())]),
+                    timestamp: SystemTime::UNIX_EPOCH,
+                    ..Event::default()
+                },
+                identity,
+                unbounded_limits,
+            );
+            serialized_event_len(&event)
+                .saturating_add(MAX_SERIALIZED_TIMESTAMP_BYTES.saturating_sub(1))
+        })
+        .max()
+        .unwrap_or(usize::MAX)
+}
+
 fn serialized_event_len(event: &Event<'_>) -> usize {
     serde_json::to_vec(event).map_or(usize::MAX, |serialized| serialized.len())
 }
@@ -223,10 +281,12 @@ fn serialized_event_len(event: &Event<'_>) -> usize {
 mod tests {
     use std::{
         collections::BTreeMap,
+        panic::AssertUnwindSafe,
         sync::{
             Arc,
             atomic::{AtomicUsize, Ordering},
         },
+        time::Duration,
     };
 
     use opentelemetry::trace::TracerProvider as _;
@@ -237,7 +297,8 @@ mod tests {
     };
 
     use super::{
-        CriticalErrorCategory, capture_critical_error, client_options, sanitize_event,
+        CriticalErrorCategory, capture_critical_error, client_options,
+        required_critical_event_budget, sanitize_event, validate_critical_event_budget,
         with_request_correlation,
     };
     use crate::{
@@ -395,6 +456,14 @@ mod tests {
             event.tags.get("category").map(String::as_str),
             Some("data_integrity")
         );
+        assert_eq!(
+            event
+                .fingerprint
+                .iter()
+                .map(AsRef::as_ref)
+                .collect::<Vec<&str>>(),
+            vec!["admin-api", "data_integrity"]
+        );
         assert_eq!(
             event.tags.get("request_id").map(String::as_str),
             Some("request-123")
@@ -409,11 +478,33 @@ mod tests {
     fn panic_creates_exactly_one_sanitized_critical_event() {
         let options =
             sentry::apply_defaults(client_options(identity(), RedactionLimits::default()));
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .build()
+            .expect("runtime");
+        let provider = SdkTracerProvider::builder().build();
+        let tracer = provider.tracer("panic-correlation-test");
+        let subscriber = build_subscriber_with_tracer(
+            ObservabilityConfig::new(identity(), "info", RedactionLimits::default()),
+            std::io::sink,
+            Some(tracer),
+        )
+        .expect("subscriber");
+        let dispatch = tracing::Dispatch::new(subscriber);
         let events = sentry::test::with_captured_events_options(
             || {
-                let result = std::panic::catch_unwind(|| {
-                    panic!("password=must-not-leak");
-                });
+                let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
+                    tracing::dispatcher::with_default(&dispatch, || {
+                        runtime.block_on(with_request_correlation(
+                            "panic-request-123".to_owned(),
+                            async {
+                                let span =
+                                    tracing::info_span!(target: "crank::trace", "http.request");
+                                let _span_guard = span.enter();
+                                panic!("password=must-not-leak");
+                            },
+                        ));
+                    });
+                }));
                 assert!(result.is_err());
             },
             options,
@@ -425,24 +516,34 @@ mod tests {
             event.tags.get("category").map(String::as_str),
             Some("panic")
         );
+        assert_eq!(
+            event.tags.get("request_id").map(String::as_str),
+            Some("panic-request-123")
+        );
+        assert_eq!(event.tags.get("trace_id").map(String::len), Some(32));
         let serialized = serde_json::to_string(event).expect("serialize event");
         assert!(!serialized.contains("must-not-leak"));
         assert!(!serialized.contains("password"));
+        provider.shutdown().expect("provider shutdown");
     }
 
     #[test]
     fn receiver_failure_does_not_change_product_result_or_recurse() {
-        struct DroppingTransport {
+        struct UnavailableTransport {
             attempts: AtomicUsize,
         }
 
-        impl sentry::Transport for DroppingTransport {
+        impl sentry::Transport for UnavailableTransport {
             fn send_envelope(&self, _envelope: Envelope) {
                 self.attempts.fetch_add(1, Ordering::Relaxed);
             }
+
+            fn flush(&self, _timeout: Duration) -> bool {
+                false
+            }
         }
 
-        let transport = Arc::new(DroppingTransport {
+        let transport = Arc::new(UnavailableTransport {
             attempts: AtomicUsize::new(0),
         });
         let mut options =
@@ -454,7 +555,10 @@ mod tests {
         );
         options.transport = Some(Arc::new(transport.clone()));
         let client = Arc::new(sentry::Client::from(options));
-        let hub = Arc::new(Hub::new(Some(client), Arc::new(Default::default())));
+        let hub = Arc::new(Hub::new(
+            Some(Arc::clone(&client)),
+            Arc::new(Default::default()),
+        ));
 
         let product_result = Hub::run(hub, || {
             capture_critical_error(CriticalErrorCategory::Internal);
@@ -463,5 +567,58 @@ mod tests {
 
         assert_eq!(product_result, 42);
         assert_eq!(transport.attempts.load(Ordering::Relaxed), 1);
+        assert!(!client.close(Some(Duration::from_millis(10))));
+        assert_eq!(transport.attempts.load(Ordering::Relaxed), 1);
+    }
+
+    #[test]
+    fn critical_event_budget_rejects_identity_that_does_not_fit() {
+        let maximum_label = "a".repeat(64);
+        let identity =
+            ServiceIdentity::try_new(maximum_label.clone(), maximum_label.clone(), maximum_label)
+                .expect("maximum identity");
+        let limits = RedactionLimits {
+            max_event_bytes: 512,
+            ..RedactionLimits::default()
+        };
+
+        assert!(matches!(
+            validate_critical_event_budget(&identity, limits),
+            Err(super::SentryConfigError::EventBudgetTooSmall)
+        ));
+        validate_critical_event_budget(
+            &identity,
+            RedactionLimits {
+                max_event_bytes: 1024,
+                ..limits
+            },
+        )
+        .expect("larger budget must hold the required identity");
+    }
+
+    #[test]
+    fn critical_event_budget_covers_every_category() {
+        let identity = identity();
+        let limits = RedactionLimits::default();
+        let required_budget = required_critical_event_budget(&identity, limits);
+
+        validate_critical_event_budget(
+            &identity,
+            RedactionLimits {
+                max_event_bytes: required_budget,
+                ..limits
+            },
+        )
+        .expect("exact required budget must be accepted");
+        assert!(matches!(
+            validate_critical_event_budget(
+                &identity,
+                RedactionLimits {
+                    max_event_bytes: required_budget - 1,
+                    ..limits
+                },
+            ),
+            Err(super::SentryConfigError::EventBudgetTooSmall)
+        ));
     }
 }

crates/crank-observability/src/incidents.rs

@@ -14,13 +14,11 @@ pub fn record_operational_incident(incident: OperationalIncident) {
     });
     match incident {
         OperationalIncident::InvocationHistoryLost => {
-            metrics::counter!("crank_invocation_history_lost_total").increment(1);
-            metrics::counter!(
-                "crank_telemetry_export_failures_total",
-                "signal_type" => "invocation_history",
-                "exporter" => "postgres"
-            )
-            .increment(1);
+            crank_metrics::record_invocation_history_lost();
+            crank_metrics::record_export_failure(
+                crank_metrics::SignalType::InvocationHistory,
+                crank_metrics::Exporter::Postgres,
+            );
         }
     }
 }

crates/crank-observability/src/instrumentation.rs

@@ -5,7 +5,8 @@ use axum::{
     middleware::Next,
     response::Response,
 };
-use metrics::{Gauge, Unit};
+use crank_metrics::{DbPoolState, HttpMethod, HttpRoute, HttpStatusClass, InFlightGuard};
+use metrics::Unit;
 
 use crate::{MetricKind, MetricUnit, metric_schema};
 
@@ -13,36 +14,28 @@ pub async fn record_http_request(request: Request, next: Next) -> Response {
     let route = request
         .extensions()
         .get::<MatchedPath>()
-        .map_or("unmatched", MatchedPath::as_str)
-        .to_owned();
-    let method = normalized_http_method(request.method().as_str());
+        .map_or(HttpRoute::unmatched(), |path| {
+            HttpRoute::from_matched_path(path.as_str())
+        });
+    let method = HttpMethod::classify(request.method().as_str());
     let started_at = Instant::now();
-    let _inflight = GaugeGuard::increment("crank_http_inflight");
+    let _inflight = InFlightGuard::http();
 
     let response = next.run(request).await;
-    let status_class = status_class(response.status().as_u16());
-
-    metrics::counter!(
-        "crank_http_requests_total",
-        "route" => route.clone(),
-        "method" => method,
-        "status_class" => status_class
-    )
-    .increment(1);
-    metrics::histogram!(
-        "crank_http_request_duration_seconds",
-        "route" => route,
-        "method" => method
-    )
-    .record(started_at.elapsed().as_secs_f64());
+    crank_metrics::record_http_request(
+        route,
+        method,
+        HttpStatusClass::from_status(response.status().as_u16()),
+        started_at.elapsed(),
+    );
 
     response
 }
 
 pub fn record_db_pool_connections(total: u32, idle: usize) {
-    let idle = idle.min(total as usize) as f64;
-    metrics::gauge!("crank_db_pool_connections", "state" => "idle").set(idle);
-    metrics::gauge!("crank_db_pool_connections", "state" => "used").set(f64::from(total) - idle);
+    let idle = u32::try_from(idle.min(total as usize)).unwrap_or(total);
+    crank_metrics::set_db_pool_connections(DbPoolState::Idle, idle);
+    crank_metrics::set_db_pool_connections(DbPoolState::Used, total.saturating_sub(idle));
 }
 
 pub(crate) fn register_metric_schema() {
@@ -64,70 +57,29 @@ pub(crate) fn register_metric_schema() {
         }
     }
 
-    metrics::gauge!("crank_http_inflight").set(0.0);
-    metrics::gauge!("crank_mcp_active_sessions").set(0.0);
-    metrics::gauge!("crank_runtime_inflight").set(0.0);
-    metrics::gauge!("crank_db_pool_connections", "state" => "idle").set(0.0);
-    metrics::gauge!("crank_db_pool_connections", "state" => "used").set(0.0);
-    metrics::gauge!("crank_catalog_tools").set(0.0);
-    metrics::gauge!("crank_catalog_estimated_context_tokens").set(0.0);
-    metrics::gauge!("crank_catalog_warnings").set(0.0);
-}
-
-fn normalized_http_method(method: &str) -> &'static str {
-    match method {
-        "GET" => "GET",
-        "POST" => "POST",
-        "PUT" => "PUT",
-        "PATCH" => "PATCH",
-        "DELETE" => "DELETE",
-        "OPTIONS" => "OPTIONS",
-        "HEAD" => "HEAD",
-        "CONNECT" => "CONNECT",
-        "TRACE" => "TRACE",
-        _ => "OTHER",
-    }
-}
-
-fn status_class(status: u16) -> &'static str {
-    match status {
-        100..=199 => "1xx",
-        200..=299 => "2xx",
-        300..=399 => "3xx",
-        400..=499 => "4xx",
-        500..=599 => "5xx",
-        _ => "other",
-    }
-}
-
-struct GaugeGuard {
-    gauge: Gauge,
-}
-
-impl GaugeGuard {
-    fn increment(name: &'static str) -> Self {
-        let gauge = metrics::gauge!(name);
-        gauge.increment(1.0);
-        Self { gauge }
-    }
-}
-
-impl Drop for GaugeGuard {
-    fn drop(&mut self) {
-        self.gauge.decrement(1.0);
-    }
+    crank_metrics::initialize_gauges();
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{normalized_http_method, status_class};
+    use crank_metrics::{HttpMethod, HttpRoute, HttpStatusClass};
 
     #[test]
     fn normalizes_unbounded_http_values() {
-        assert_eq!(normalized_http_method("GET"), "GET");
-        assert_eq!(normalized_http_method("CUSTOM-user-controlled"), "OTHER");
-        assert_eq!(status_class(204), "2xx");
-        assert_eq!(status_class(429), "4xx");
-        assert_eq!(status_class(999), "other");
+        assert_eq!(HttpMethod::classify("GET"), HttpMethod::Get);
+        assert_eq!(
+            HttpMethod::classify("CUSTOM-user-controlled"),
+            HttpMethod::Other
+        );
+        assert_eq!(HttpStatusClass::from_status(204), HttpStatusClass::Success);
+        assert_eq!(
+            HttpStatusClass::from_status(429),
+            HttpStatusClass::ClientError
+        );
+        assert_eq!(HttpStatusClass::from_status(999), HttpStatusClass::Other);
+        assert_eq!(
+            HttpRoute::from_matched_path("/documents/{document_id}"),
+            HttpRoute::unmatched()
+        );
     }
 }

crates/crank-observability/src/lib.rs

@@ -5,7 +5,6 @@ mod incidents;
 mod instrumentation;
 mod lifecycle;
 mod logging;
-mod metrics_schema;
 mod otlp;
 mod prometheus;
 mod propagation;
@@ -14,6 +13,9 @@ mod schema;
 
 pub use config::{ObservabilityConfig, ObservabilityConfigError, ServiceIdentity};
 pub use correlation::RequestId;
+pub use crank_metrics::{
+    DURATION_BUCKETS_SECONDS, MetricDefinition, MetricKind, MetricUnit, metric_schema,
+};
 pub use error_reporting::{
     CriticalErrorCategory, SentryConfig, SentryConfigError, capture_critical_error,
     with_request_correlation,
@@ -22,9 +24,6 @@ pub use incidents::{OperationalIncident, operational_incident_total, record_oper
 pub use instrumentation::{record_db_pool_connections, record_http_request};
 pub use lifecycle::{ObservabilityInitError, ObservabilityLifecycle, init};
 pub use logging::build_subscriber;
-pub use metrics_schema::{
-    DURATION_BUCKETS_SECONDS, MetricDefinition, MetricKind, MetricUnit, metric_schema,
-};
 pub use otlp::{
     OtlpBatchConfig, OtlpTraceConfig, OtlpTraceConfigError, OtlpTraceError, build_tracer_provider,
 };

crates/crank-observability/src/lifecycle.rs

@@ -33,7 +33,7 @@ impl ObservabilityLifecycle {
             .map_err(|_| ObservabilityInitError::SubscriberAlreadyInitialized)?;
         let metrics_handle = install_prometheus_recorder(&identity)?;
         register_metric_schema();
-        let sentry_guard = init_sentry(&identity, redaction_limits, sentry_config);
+        let sentry_guard = init_sentry(&identity, redaction_limits, sentry_config)?;
 
         Ok(Self {
             metrics_handle,

crates/crank-observability/src/otlp.rs

@@ -340,12 +340,10 @@ impl SpanExporterTrait for ObservedSpanExporter {
         sanitize_trace_batch(&mut batch);
         let result = self.0.export(batch).await;
         if result.is_err() {
-            metrics::counter!(
-                "crank_telemetry_export_failures_total",
-                "signal_type" => "trace",
-                "exporter" => "otlp"
-            )
-            .increment(1);
+            crank_metrics::record_export_failure(
+                crank_metrics::SignalType::Trace,
+                crank_metrics::Exporter::Otlp,
+            );
         }
         result
     }

crates/crank-observability/src/prometheus.rs

@@ -155,7 +155,7 @@ impl MetricsSurface {
         Router::new()
             .route("/metrics", get(render_metrics))
             .route("/health", get(metrics_health))
-            .layer(middleware::from_fn_with_state(
+            .route_layer(middleware::from_fn_with_state(
                 self.state.clone(),
                 authorize_metrics,
             ))
@@ -179,6 +179,12 @@ pub struct MetricsServer {
 }
 
 impl MetricsServer {
+    pub fn local_addr(&self) -> Result<SocketAddr, MetricsServeError> {
+        self.listener
+            .local_addr()
+            .map_err(|_| MetricsServeError::LocalAddress)
+    }
+
     pub async fn serve(self) -> Result<(), MetricsServeError> {
         axum::serve(self.listener, self.router)
             .await
@@ -198,6 +204,8 @@ pub enum MetricsServeError {
     Bind,
     #[error("metrics listener stopped unexpectedly")]
     Serve,
+    #[error("failed to read metrics listener address")]
+    LocalAddress,
 }
 
 pub(crate) fn install_prometheus_recorder(
@@ -257,10 +265,15 @@ async fn authorize_metrics(
 }
 
 fn bearer_token(headers: &HeaderMap) -> Option<&[u8]> {
-    headers
-        .get(header::AUTHORIZATION)?
-        .as_bytes()
-        .strip_prefix(b"Bearer ")
+    let value = headers.get(header::AUTHORIZATION)?.as_bytes();
+    let separator = value.iter().position(|byte| *byte == b' ')?;
+    let (scheme, token_with_spaces) = value.split_at(separator);
+    let token_start = token_with_spaces.iter().position(|byte| *byte != b' ')?;
+    let token = &token_with_spaces[token_start..];
+
+    scheme
+        .eq_ignore_ascii_case(b"bearer")
+        .then_some(token)
         .filter(|token| !token.is_empty())
 }
 

crates/crank-observability/tests/http_metrics.rs

@@ -31,19 +31,21 @@ async fn http_metrics_use_matched_routes_and_closed_labels() {
     let metrics = lifecycle.metrics_surface(config).router();
     let app = Router::new()
         .route(
-            "/documents/{document_id}",
+            "/api/admin/workspaces/{workspace_id}/operations/{operation_id}",
             get(|| async { StatusCode::NO_CONTENT }),
         )
         .layer(middleware::from_fn(record_http_request));
 
-    let sensitive_path_segment = "customer-secret-document-id";
+    let sensitive_path_segment = "customer-secret-operation-id";
     for index in 0..100 {
         let response = app
             .clone()
             .oneshot(
-                Request::get(format!("/documents/{sensitive_path_segment}-{index}"))
-                    .body(Body::empty())
-                    .expect("request"),
+                Request::get(format!(
+                    "/api/admin/workspaces/customer-secret-workspace-{index}/operations/{sensitive_path_segment}-{index}"
+                ))
+                .body(Body::empty())
+                .expect("request"),
             )
             .await
             .expect("response");
@@ -74,7 +76,9 @@ async fn http_metrics_use_matched_routes_and_closed_labels() {
     let body = String::from_utf8(body.to_vec()).expect("utf-8 metrics");
 
     assert!(body.contains("crank_http_requests_total"));
-    assert!(body.contains("route=\"/documents/{document_id}\""));
+    assert!(
+        body.contains("route=\"/api/admin/workspaces/{workspace_id}/operations/{operation_id}\"")
+    );
     assert!(body.contains("method=\"GET\""));
     assert!(body.contains("status_class=\"2xx\""));
     assert!(body.contains("crank_http_request_duration_seconds_bucket"));
@@ -83,7 +87,9 @@ async fn http_metrics_use_matched_routes_and_closed_labels() {
         body.lines()
             .filter(|line| {
                 line.starts_with("crank_http_requests_total{")
-                    && line.contains("route=\"/documents/{document_id}\"")
+                    && line.contains(
+                        "route=\"/api/admin/workspaces/{workspace_id}/operations/{operation_id}\"",
+                    )
             })
             .count(),
         1,

crates/crank-observability/tests/metrics_signals.rs

@@ -0,0 +1,134 @@
+use std::{
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    time::Duration,
+};
+
+use crank_metrics::{
+    CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute,
+    HttpStatusClass, IdempotencyOutcome, InFlightGuard, InvocationSource, LimitStage, McpMethod,
+    McpOutcome, McpResponseMode, SignalType, ToolErrorKind, ToolOutcome, UpstreamOperationKind,
+    UpstreamOutcome, record_cache_outcome, record_confirmation_outcome, record_export_failure,
+    record_http_request, record_idempotency_outcome, record_invocation_history_lost,
+    record_limit_rejection, record_mcp_request, record_tool_invocation, record_upstream_request,
+    set_catalog, set_db_pool_connections, set_mcp_active_sessions,
+};
+use crank_observability::{MetricsConfig, ObservabilityConfig, RedactionLimits, ServiceIdentity};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+#[tokio::test]
+async fn typed_product_signals_are_exposed_by_a_real_scrape() {
+    let identity =
+        ServiceIdentity::try_new("metrics-signals", "0.3.1", "test").expect("valid identity");
+    let lifecycle = crank_observability::init(ObservabilityConfig::new(
+        identity,
+        "off",
+        RedactionLimits::default(),
+    ))
+    .expect("observability lifecycle");
+    let config = MetricsConfig::new(
+        true,
+        SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0),
+        None,
+    )
+    .expect("loopback metrics config");
+    let server = lifecycle
+        .metrics_surface(config)
+        .bind()
+        .await
+        .expect("metrics listener");
+    let address = server.local_addr().expect("listener address");
+    let task = tokio::spawn(server.serve());
+
+    record_http_request(
+        HttpRoute::from_matched_path("/api/auth/session"),
+        HttpMethod::Get,
+        HttpStatusClass::Success,
+        Duration::from_millis(7),
+    );
+    record_mcp_request(
+        McpMethod::ToolsCall,
+        McpResponseMode::Json,
+        McpOutcome::ToolError,
+    );
+    set_mcp_active_sessions(2);
+    let _stream = InFlightGuard::mcp_stream();
+    let _runtime = InFlightGuard::runtime();
+    record_tool_invocation(
+        InvocationSource::AgentToolCall,
+        ToolOutcome::Error,
+        ToolErrorKind::Mapping,
+        Duration::from_millis(11),
+    );
+    record_upstream_request(
+        UpstreamOperationKind::Rest,
+        UpstreamOutcome::Timeout,
+        Duration::from_millis(13),
+    );
+    record_limit_rejection(LimitStage::Concurrency);
+    record_cache_outcome(CacheOutcome::Hit);
+    record_idempotency_outcome(IdempotencyOutcome::Replay);
+    record_confirmation_outcome(ConfirmationOutcome::Required);
+    set_db_pool_connections(DbPoolState::Idle, 3);
+    set_db_pool_connections(DbPoolState::Used, 1);
+    set_catalog(5, 1200, 1);
+    record_invocation_history_lost();
+    record_export_failure(SignalType::Trace, Exporter::Otlp);
+
+    let response = raw_get(address, "/metrics").await;
+    task.abort();
+    let _ = task.await;
+
+    assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}");
+    for metric in [
+        "crank_http_requests_total",
+        "crank_mcp_requests_total",
+        "crank_mcp_active_sessions",
+        "crank_mcp_active_streams",
+        "crank_tool_invocations_total",
+        "crank_upstream_requests_total",
+        "crank_runtime_inflight",
+        "crank_runtime_limit_rejections_total",
+        "crank_runtime_cache_total",
+        "crank_idempotency_total",
+        "crank_confirmation_total",
+        "crank_db_pool_connections",
+        "crank_catalog_tools",
+        "crank_invocation_history_lost_total",
+        "crank_telemetry_export_failures_total",
+    ] {
+        assert!(response.contains(metric), "missing metric {metric}");
+    }
+    for expected_label in [
+        "outcome=\"tool_error\"",
+        "outcome=\"timeout\"",
+        "outcome=\"hit\"",
+        "outcome=\"replay\"",
+        "outcome=\"required\"",
+        "error_kind=\"mapping\"",
+    ] {
+        assert!(
+            response.contains(expected_label),
+            "missing label {expected_label}"
+        );
+    }
+    assert!(!response.contains("customer-secret"));
+}
+
+async fn raw_get(address: SocketAddr, path: &str) -> String {
+    let mut stream = tokio::net::TcpStream::connect(address)
+        .await
+        .expect("metrics listener connection");
+    stream
+        .write_all(
+            format!("GET {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n")
+                .as_bytes(),
+        )
+        .await
+        .expect("metrics request");
+    let mut response = Vec::new();
+    stream
+        .read_to_end(&mut response)
+        .await
+        .expect("metrics response");
+    String::from_utf8(response).expect("utf-8 response")
+}

crates/crank-observability/tests/prometheus.rs

@@ -8,6 +8,7 @@ use crank_observability::{
     DURATION_BUCKETS_SECONDS, MetricsConfig, MetricsConfigError, MetricsSurface, ServiceIdentity,
     metric_schema,
 };
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tower::ServiceExt;
 
 fn identity() -> ServiceIdentity {
@@ -27,6 +28,33 @@ fn loopback_is_allowed_without_a_token() {
     assert!(!config.requires_authentication());
 }
 
+#[tokio::test]
+async fn default_service_ports_bind_and_serve_real_metrics_listeners() {
+    for (service, port) in [("admin-api", 9464), ("mcp-server", 9465)] {
+        let config = MetricsConfig::new(
+            true,
+            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
+            None,
+        )
+        .expect("default loopback metrics config");
+        let identity =
+            ServiceIdentity::try_new(service, "0.3.1", "test").expect("valid service identity");
+        let server = MetricsSurface::for_test(config, identity)
+            .expect("metrics surface")
+            .bind()
+            .await
+            .expect("default metrics port must bind");
+        assert_eq!(server.local_addr().expect("local address").port(), port);
+
+        let task = tokio::spawn(server.serve());
+        let response = raw_get(port, "/metrics").await;
+        assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}");
+        assert!(response.contains("content-type: text/plain"));
+        task.abort();
+        let _ = task.await;
+    }
+}
+
 #[test]
 fn non_loopback_without_a_token_is_rejected_without_secret_data() {
     let error = MetricsConfig::new(
@@ -51,8 +79,13 @@ fn schema_is_closed_and_uses_fixed_duration_buckets() {
     assert!(names.contains(&"crank_http_requests_total"));
     assert!(names.contains(&"crank_http_request_duration_seconds"));
     assert!(names.contains(&"crank_mcp_requests_total"));
+    assert!(names.contains(&"crank_mcp_active_sessions"));
+    assert!(names.contains(&"crank_mcp_active_streams"));
     assert!(names.contains(&"crank_tool_invocations_total"));
     assert!(names.contains(&"crank_runtime_inflight"));
+    assert!(names.contains(&"crank_runtime_cache_total"));
+    assert!(names.contains(&"crank_idempotency_total"));
+    assert!(names.contains(&"crank_confirmation_total"));
     assert!(names.contains(&"crank_db_pool_connections"));
     assert!(names.contains(&"crank_catalog_tools"));
     assert!(names.contains(&"crank_invocation_history_lost_total"));
@@ -108,7 +141,7 @@ async fn external_surface_protects_both_routes_and_exposes_nothing_else() {
             .clone()
             .oneshot(
                 Request::get(path)
-                    .header(header::AUTHORIZATION, format!("Bearer {token}"))
+                    .header(header::AUTHORIZATION, format!("bEaReR  {token}"))
                     .body(Body::empty())
                     .expect("request"),
             )
@@ -121,14 +154,35 @@ async fn external_surface_protects_both_routes_and_exposes_nothing_else() {
         assert!(!String::from_utf8_lossy(&body).contains(token));
     }
 
-    let absent = app
-        .oneshot(
-            Request::get("/api/operations")
-                .header(header::AUTHORIZATION, format!("Bearer {token}"))
-                .body(Body::empty())
-                .expect("request"),
+    for authorization in [None, Some(format!("Bearer {token}"))] {
+        let mut request = Request::get("/api/operations");
+        if let Some(authorization) = authorization {
+            request = request.header(header::AUTHORIZATION, authorization);
+        }
+        let absent = app
+            .clone()
+            .oneshot(request.body(Body::empty()).expect("request"))
+            .await
+            .expect("response");
+        assert_eq!(absent.status(), StatusCode::NOT_FOUND);
+    }
+}
+
+async fn raw_get(port: u16, path: &str) -> String {
+    let mut stream = tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, port))
+        .await
+        .expect("metrics listener connection");
+    stream
+        .write_all(
+            format!("GET {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nConnection: close\r\n\r\n")
+                .as_bytes(),
         )
         .await
-        .expect("response");
-    assert_eq!(absent.status(), StatusCode::NOT_FOUND);
+        .expect("metrics request");
+    let mut response = Vec::new();
+    stream
+        .read_to_end(&mut response)
+        .await
+        .expect("metrics response");
+    String::from_utf8(response).expect("utf-8 response")
 }

crates/crank-runtime/Cargo.toml

@@ -19,10 +19,10 @@ base64.workspace = true
 crank-adapter-rest = { path = "../crank-adapter-rest" }
 crank-core = { path = "../crank-core" }
 crank-mapping = { path = "../crank-mapping" }
+crank-metrics = { path = "../crank-metrics" }
 crank-schema = { path = "../crank-schema" }
 crank-trace = { path = "../crank-trace" }
 hkdf.workspace = true
-metrics.workspace = true
 redis = { version = "0.29", features = ["tokio-comp", "connection-manager"] }
 serde.workspace = true
 serde_json.workspace = true
@@ -36,6 +36,8 @@ uuid.workspace = true
 [dev-dependencies]
 axum.workspace = true
 futures-util = "0.3"
+metrics.workspace = true
+metrics-util = "0.20.4"
 testcontainers.workspace = true
 time.workspace = true
 tracing-subscriber.workspace = true

crates/crank-runtime/src/confirmation.rs

@@ -38,7 +38,14 @@ pub async fn confirm_operation(
         .or_else(|| confirmation_token_from_input(input));
 
     let Some(provided_token) = provided_token else {
-        let token = issue_confirmation_token(store, &scope, &input_hash, &safety).await?;
+        let token = issue_confirmation_token(
+            store,
+            operation.operation_id.as_str(),
+            &scope,
+            &input_hash,
+            &safety,
+        )
+        .await?;
         return Err(RuntimeError::ConfirmationRequired {
             operation_id: operation.operation_id.as_str().to_owned(),
             safety_class: safety.class,
@@ -47,7 +54,14 @@ pub async fn confirm_operation(
         });
     };
 
-    consume_confirmation_token(store, &scope, provided_token, &input_hash).await
+    consume_confirmation_token(
+        store,
+        operation.operation_id.as_str(),
+        &scope,
+        provided_token,
+        &input_hash,
+    )
+    .await
 }
 
 pub(crate) fn is_applicable(operation: &RuntimeOperation) -> bool {
@@ -108,6 +122,7 @@ fn confirmation_scope(
 
 async fn issue_confirmation_token(
     store: &dyn CoordinationStateStore,
+    operation_id: &str,
     scope: &str,
     input_hash: &str,
     safety: &OperationSafetyPolicy,
@@ -128,15 +143,15 @@ async fn issue_confirmation_token(
             Duration::from_millis(ttl_ms),
         )
         .await
-        .map_err(|error| RuntimeError::InvalidPreparedRequest {
-            field: "confirmation_token".to_owned(),
-            reason: error.to_string(),
+        .map_err(|_| RuntimeError::ConfirmationStoreUnavailable {
+            operation_id: operation_id.to_owned(),
         })?;
     Ok(token)
 }
 
 async fn consume_confirmation_token(
     store: &dyn CoordinationStateStore,
+    operation_id: &str,
     operation_scope: &str,
     token: &str,
     input_hash: &str,
@@ -145,9 +160,8 @@ async fn consume_confirmation_token(
     let stored = store
         .take_value(CacheScope::Coordination, &key)
         .await
-        .map_err(|error| RuntimeError::InvalidPreparedRequest {
-            field: "confirmation_token".to_owned(),
-            reason: error.to_string(),
+        .map_err(|_| RuntimeError::ConfirmationStoreUnavailable {
+            operation_id: operation_id.to_owned(),
         })?;
 
     let Some(stored) = stored else {

crates/crank-runtime/src/executor.rs

@@ -5,12 +5,17 @@ use crank_core::{
     AdapterRegistry, CoordinationStateStore, ExecutionMode, InvocationSource, InvocationStatus,
     MeteringEvent, ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter,
 };
+use crank_metrics::{
+    CacheOutcome, ConfirmationOutcome, IdempotencyOutcome, InFlightGuard,
+    InvocationSource as MetricInvocationSource, LimitStage, ToolErrorKind, ToolInvocationMetrics,
+    ToolOutcome, record_cache_outcome, record_confirmation_outcome, record_idempotency_outcome,
+    record_limit_rejection,
+};
 use crank_trace::{ErrorCategory, Stage, StageOutcome};
-use metrics::Gauge;
 use serde_json::{Map, Value, json};
 use time::OffsetDateTime;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
-use tracing::{Instrument, Span, debug};
+use tracing::{Instrument, Span, debug, warn};
 use uuid::Uuid;
 
 use crate::{
@@ -42,6 +47,65 @@ pub struct RuntimeExecutionRequest<'a> {
     pub request_context: Option<&'a RuntimeRequestContext>,
 }
 
+struct IdempotencyCancellationGuard {
+    cleanup: Option<IdempotencyCancellationCleanup>,
+    runtime: tokio::runtime::Handle,
+}
+
+struct IdempotencyCancellationCleanup {
+    store: Arc<dyn CoordinationStateStore>,
+    operation: RuntimeOperation,
+    reservation: crate::idempotency::IdempotencyReservation,
+}
+
+impl IdempotencyCancellationGuard {
+    fn new(
+        store: Arc<dyn CoordinationStateStore>,
+        operation: RuntimeOperation,
+        reservation: crate::idempotency::IdempotencyReservation,
+    ) -> Self {
+        Self {
+            cleanup: Some(IdempotencyCancellationCleanup {
+                store,
+                operation,
+                reservation,
+            }),
+            runtime: tokio::runtime::Handle::current(),
+        }
+    }
+
+    fn disarm(&mut self) {
+        self.cleanup = None;
+    }
+}
+
+impl Drop for IdempotencyCancellationGuard {
+    fn drop(&mut self) {
+        let Some(cleanup) = self.cleanup.take() else {
+            return;
+        };
+        self.runtime.spawn(async move {
+            let result = crate::idempotency::mark_outcome_unknown(
+                cleanup.store.as_ref(),
+                &cleanup.operation,
+                &cleanup.reservation,
+            )
+            .await;
+            record_idempotency_outcome(match &result {
+                Ok(()) => IdempotencyOutcome::OutcomeUnknown,
+                Err(error) => idempotency_error_outcome(error),
+            });
+            if result.is_err() {
+                warn!(
+                    name: "runtime.idempotency.cancellation_cleanup_failed",
+                    error_category = "idempotency_store",
+                    "failed to finalize cancelled idempotent execution"
+                );
+            }
+        });
+    }
+}
+
 impl<'a> RuntimeExecutionRequest<'a> {
     pub fn new(operation: &'a RuntimeOperation, input: &'a Value) -> Self {
         Self {
@@ -180,10 +244,12 @@ impl RuntimeExecutor {
     ) -> Result<Value, RuntimeError> {
         log_runtime_event("unary.execute", request.operation, request.request_context);
         let started_at = Instant::now();
+        let invocation_metrics =
+            ToolInvocationMetrics::start(metric_invocation_source(request.request_context));
         let runtime_span = Stage::RuntimeExecute.span();
         let result = async {
             let _permit = self.acquire_unary_permit(request.operation)?;
-            let _inflight = RuntimeInFlightGuard::new();
+            let _inflight = InFlightGuard::runtime();
             let mapping_span = Stage::RuntimeArgumentsMap.span();
             let prepared_request =
                 mapping_span.in_scope(|| self.prepare_request(request.operation, request.input));
@@ -203,7 +269,11 @@ impl RuntimeExecutor {
         .await;
         record_runtime_result(&runtime_span, &result);
         drop(runtime_span);
-        record_execution_metrics(request.request_context, &result, started_at);
+        let (outcome, error_kind) = match &result {
+            Ok(_) => (ToolOutcome::Success, ToolErrorKind::None),
+            Err(error) => (ToolOutcome::Error, runtime_error_kind(error)),
+        };
+        invocation_metrics.complete(outcome, error_kind);
         self.record_metering(
             request.operation,
             request.request_context,
@@ -245,6 +315,7 @@ impl RuntimeExecutor {
         ) {
             Ok(key) => key,
             Err(error) if idempotency_applicable => {
+                record_idempotency_outcome(idempotency_error_outcome(&error));
                 let span = Stage::RuntimeIdempotency.span();
                 StageOutcome::Error.record(&span);
                 ErrorCategory::Idempotency.record(&span);
@@ -264,14 +335,19 @@ impl RuntimeExecutor {
             .instrument(approval_span.clone())
             .await;
             match &approval_result {
-                Ok(()) => StageOutcome::Success.record(&approval_span),
+                Ok(()) => {
+                    StageOutcome::Success.record(&approval_span);
+                    record_confirmation_outcome(ConfirmationOutcome::Approved);
+                }
                 Err(RuntimeError::ConfirmationRequired { .. }) => {
                     StageOutcome::Required.record(&approval_span);
                     ErrorCategory::Approval.record(&approval_span);
+                    record_confirmation_outcome(ConfirmationOutcome::Required);
                 }
                 Err(error) => {
                     StageOutcome::Error.record(&approval_span);
                     runtime_error_category(error).record(&approval_span);
+                    record_confirmation_outcome(confirmation_error_outcome(error));
                 }
             }
             drop(approval_span);
@@ -292,9 +368,11 @@ impl RuntimeExecutor {
             match &result {
                 Ok(crate::idempotency::IdempotencyAction::Execute(_)) => {
                     StageOutcome::Execute.record(&idempotency_span);
+                    record_idempotency_outcome(IdempotencyOutcome::Execute);
                 }
                 Ok(crate::idempotency::IdempotencyAction::Replay(_)) => {
                     StageOutcome::Replay.record(&idempotency_span);
+                    record_idempotency_outcome(IdempotencyOutcome::Replay);
                 }
                 Ok(crate::idempotency::IdempotencyAction::Disabled) => {
                     StageOutcome::Skipped.record(&idempotency_span);
@@ -302,6 +380,7 @@ impl RuntimeExecutor {
                 Err(error) => {
                     StageOutcome::Error.record(&idempotency_span);
                     runtime_error_category(error).record(&idempotency_span);
+                    record_idempotency_outcome(idempotency_error_outcome(error));
                 }
             }
             drop(idempotency_span);
@@ -312,6 +391,18 @@ impl RuntimeExecutor {
         if let crate::idempotency::IdempotencyAction::Replay(response) = &idempotency {
             return transform_response(operation, response);
         }
+        let mut cancellation_guard =
+            if let crate::idempotency::IdempotencyAction::Execute(reservation) = &idempotency {
+                self.coordination_store.as_ref().map(|store| {
+                    IdempotencyCancellationGuard::new(
+                        Arc::clone(store),
+                        operation.clone(),
+                        reservation.clone(),
+                    )
+                })
+            } else {
+                None
+            };
 
         let adapter_result = match self
             .load_cached_adapter_response(operation, &prepared_request, request_context)
@@ -346,6 +437,13 @@ impl RuntimeExecutor {
                             .instrument(idempotency_span.clone())
                             .await;
                     record_runtime_result(&idempotency_span, &cleanup_result);
+                    record_idempotency_outcome(match &cleanup_result {
+                        Ok(()) => IdempotencyOutcome::OutcomeUnknown,
+                        Err(error) => idempotency_error_outcome(error),
+                    });
+                    if let Some(guard) = &mut cancellation_guard {
+                        guard.disarm();
+                    }
                 }
                 return Err(error);
             }
@@ -359,6 +457,13 @@ impl RuntimeExecutor {
                     .instrument(idempotency_span.clone())
                     .await;
             record_runtime_result(&idempotency_span, &completion_result);
+            record_idempotency_outcome(match &completion_result {
+                Ok(()) => IdempotencyOutcome::Completed,
+                Err(error) => idempotency_error_outcome(error),
+            });
+            if let Some(guard) = &mut cancellation_guard {
+                guard.disarm();
+            }
             drop(idempotency_span);
             completion_result?;
         }
@@ -447,8 +552,13 @@ impl RuntimeExecutor {
         let response_cache = self.response_cache.as_ref()?;
         let cache_key = response_cache_key(operation, prepared_request, request_context)?;
         let cached = match response_cache.get(&cache_key).await {
-            Ok(cached) => cached?,
+            Ok(Some(cached)) => cached,
+            Ok(None) => {
+                record_cache_outcome(CacheOutcome::Miss);
+                return None;
+            }
             Err(_) => {
+                record_cache_outcome(CacheOutcome::ReadError);
                 debug!(
                     name: "runtime.response_cache.read_failed",
                     operation_id = operation.operation_id.as_str(),
@@ -460,15 +570,21 @@ impl RuntimeExecutor {
         };
 
         match adapter_response_from_cached(cached) {
-            Ok(response) => Some(response),
+            Ok(response) => {
+                record_cache_outcome(CacheOutcome::Hit);
+                Some(response)
+            }
             Err(_) => {
+                record_cache_outcome(CacheOutcome::DecodeError);
                 debug!(
                     name: "runtime.response_cache.decode_failed",
                     operation_id = operation.operation_id.as_str(),
                     error_category = "cached_response",
                     "cached response payload was invalid"
                 );
-                let _ = response_cache.delete(&cache_key).await;
+                if response_cache.delete(&cache_key).await.is_err() {
+                    record_cache_outcome(CacheOutcome::EvictError);
+                }
                 None
             }
         }
@@ -505,12 +621,15 @@ impl RuntimeExecutor {
             .await
             .is_err()
         {
+            record_cache_outcome(CacheOutcome::WriteError);
             debug!(
                 name: "runtime.response_cache.write_failed",
                 operation_id = operation.operation_id.as_str(),
                 error_category = "response_cache",
                 "response cache write skipped"
             );
+        } else {
+            record_cache_outcome(CacheOutcome::Stored);
         }
     }
 }
@@ -634,86 +753,64 @@ fn try_acquire_limit(
     limit: usize,
 ) -> Result<OwnedSemaphorePermit, RuntimeError> {
     limiter.try_acquire_owned().map_err(|_| {
-        metrics::counter!(
-            "crank_runtime_limit_rejections_total",
-            "stage" => "concurrency"
-        )
-        .increment(1);
+        record_limit_rejection(LimitStage::Concurrency);
         RuntimeError::ConcurrencyLimitExceeded { kind, limit }
     })
 }
 
-fn record_execution_metrics<T>(
+fn metric_invocation_source(
     request_context: Option<&RuntimeRequestContext>,
-    result: &Result<T, RuntimeError>,
-    started_at: Instant,
-) {
-    let source = request_context
+) -> MetricInvocationSource {
+    request_context
         .and_then(RuntimeRequestContext::metering_context)
-        .map_or("internal", |context| match context.source {
-            InvocationSource::AdminTestRun => "admin_test_run",
-            InvocationSource::AgentToolCall => "agent_tool_call",
-        });
-    let (outcome, error_kind) = match result {
-        Ok(_) => ("success", "none"),
-        Err(error) => ("error", runtime_error_kind(error)),
-    };
-
-    metrics::counter!(
-        "crank_tool_invocations_total",
-        "source" => source,
-        "outcome" => outcome,
-        "error_kind" => error_kind
-    )
-    .increment(1);
-    metrics::histogram!(
-        "crank_tool_invocation_duration_seconds",
-        "source" => source,
-        "outcome" => outcome
-    )
-    .record(started_at.elapsed().as_secs_f64());
+        .map_or(MetricInvocationSource::Internal, |context| {
+            match context.source {
+                InvocationSource::AdminTestRun => MetricInvocationSource::AdminTestRun,
+                InvocationSource::AgentToolCall => MetricInvocationSource::AgentToolCall,
+            }
+        })
 }
 
-fn runtime_error_kind(error: &RuntimeError) -> &'static str {
+fn runtime_error_kind(error: &RuntimeError) -> ToolErrorKind {
     match error {
-        RuntimeError::Schema(_) => "schema",
-        RuntimeError::Mapping(_) => "mapping",
-        RuntimeError::RestAdapter(_) => "rest_adapter",
-        RuntimeError::ProtocolAdapter(_) => "protocol_adapter",
-        RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol",
-        RuntimeError::UnsupportedExecutionMode { .. } => "unsupported_execution_mode",
-        RuntimeError::ConcurrencyLimitExceeded { .. } => "concurrency_limit",
-        RuntimeError::InvalidPreparedRequest { .. } => "invalid_prepared_request",
-        RuntimeError::ConfirmationRequired { .. } => "confirmation_required",
-        RuntimeError::InvalidConfirmationToken { .. } => "invalid_confirmation_token",
-        RuntimeError::ConfirmationStoreUnavailable { .. } => "confirmation_store",
-        RuntimeError::IdempotencyStoreUnavailable { .. } => "idempotency_store",
-        RuntimeError::IdempotencyInProgress { .. } => "idempotency_in_progress",
-        RuntimeError::IdempotencyConflict { .. } => "idempotency_conflict",
-        RuntimeError::IdempotencyOutcomeUnknown { .. } => "idempotency_outcome_unknown",
-        RuntimeError::MissingAuthProfile { .. } => "missing_auth_profile",
-        RuntimeError::MissingSecret { .. } => "missing_secret",
-        RuntimeError::MissingSecretVersion { .. } => "missing_secret_version",
-        RuntimeError::InvalidAuthSecretValue { .. } => "invalid_auth_secret",
-        RuntimeError::SecretCrypto { .. } => "secret_crypto",
+        RuntimeError::Schema(_) => ToolErrorKind::Schema,
+        RuntimeError::Mapping(_) => ToolErrorKind::Mapping,
+        RuntimeError::RestAdapter(_) => ToolErrorKind::RestAdapter,
+        RuntimeError::ProtocolAdapter(_) => ToolErrorKind::ProtocolAdapter,
+        RuntimeError::UnsupportedProtocol { .. } => ToolErrorKind::UnsupportedProtocol,
+        RuntimeError::UnsupportedExecutionMode { .. } => ToolErrorKind::UnsupportedExecutionMode,
+        RuntimeError::ConcurrencyLimitExceeded { .. } => ToolErrorKind::ConcurrencyLimit,
+        RuntimeError::InvalidPreparedRequest { .. } => ToolErrorKind::InvalidPreparedRequest,
+        RuntimeError::ConfirmationRequired { .. } => ToolErrorKind::ConfirmationRequired,
+        RuntimeError::InvalidConfirmationToken { .. } => ToolErrorKind::InvalidConfirmationToken,
+        RuntimeError::ConfirmationStoreUnavailable { .. } => ToolErrorKind::ConfirmationStore,
+        RuntimeError::IdempotencyStoreUnavailable { .. } => ToolErrorKind::IdempotencyStore,
+        RuntimeError::IdempotencyInProgress { .. } => ToolErrorKind::IdempotencyInProgress,
+        RuntimeError::IdempotencyConflict { .. } => ToolErrorKind::IdempotencyConflict,
+        RuntimeError::IdempotencyOutcomeUnknown { .. } => ToolErrorKind::IdempotencyOutcomeUnknown,
+        RuntimeError::MissingAuthProfile { .. } => ToolErrorKind::MissingAuthProfile,
+        RuntimeError::MissingSecret { .. } => ToolErrorKind::MissingSecret,
+        RuntimeError::MissingSecretVersion { .. } => ToolErrorKind::MissingSecretVersion,
+        RuntimeError::InvalidAuthSecretValue { .. } => ToolErrorKind::InvalidAuthSecret,
+        RuntimeError::SecretCrypto { .. } => ToolErrorKind::SecretCrypto,
     }
 }
 
-struct RuntimeInFlightGuard {
-    gauge: Gauge,
-}
-
-impl RuntimeInFlightGuard {
-    fn new() -> Self {
-        let gauge = metrics::gauge!("crank_runtime_inflight");
-        gauge.increment(1.0);
-        Self { gauge }
+fn idempotency_error_outcome(error: &RuntimeError) -> IdempotencyOutcome {
+    match error {
+        RuntimeError::IdempotencyConflict { .. } => IdempotencyOutcome::Conflict,
+        RuntimeError::IdempotencyInProgress { .. } => IdempotencyOutcome::InProgress,
+        RuntimeError::IdempotencyOutcomeUnknown { .. } => IdempotencyOutcome::OutcomeUnknown,
+        RuntimeError::IdempotencyStoreUnavailable { .. } => IdempotencyOutcome::StoreUnavailable,
+        _ => IdempotencyOutcome::Error,
     }
 }
 
-impl Drop for RuntimeInFlightGuard {
-    fn drop(&mut self) {
-        self.gauge.decrement(1.0);
+fn confirmation_error_outcome(error: &RuntimeError) -> ConfirmationOutcome {
+    match error {
+        RuntimeError::InvalidConfirmationToken { .. } => ConfirmationOutcome::InvalidToken,
+        RuntimeError::ConfirmationStoreUnavailable { .. } => ConfirmationOutcome::StoreUnavailable,
+        _ => ConfirmationOutcome::Error,
     }
 }
 

crates/crank-runtime/src/idempotency.rs

@@ -21,6 +21,7 @@ pub(crate) enum IdempotencyAction {
     Replay(AdapterResponse),
 }
 
+#[derive(Clone)]
 pub(crate) struct IdempotencyReservation {
     key: String,
     initial: CoordinationStateValue,

crates/crank-runtime/src/rate_limit.rs

@@ -5,6 +5,7 @@ use std::{
 };
 
 use crank_core::{RateLimitDecision, RateLimitStateStore};
+use crank_metrics::{LimitStage, record_limit_rejection};
 use thiserror::Error;
 use tracing::warn;
 
@@ -105,11 +106,7 @@ impl RequestRateLimiter {
             }
         };
         if matches!(result, Err(RateLimitCheckError::Rejected(_))) {
-            metrics::counter!(
-                "crank_runtime_limit_rejections_total",
-                "stage" => "rate_limit"
-            )
-            .increment(1);
+            record_limit_rejection(LimitStage::RateLimit);
         }
         result
     }

crates/crank-runtime/tests/integration/confirmation.rs

@@ -6,8 +6,9 @@ use std::sync::{
 
 use async_trait::async_trait;
 use crank_core::{
-    AdapterResponse, ConfirmationPolicy, ExecutionConfig, ExecutionMode, HttpMethod, Operation,
-    OperationId, OperationSafetyClass, OperationSafetyPolicy, OperationSecurityLevel,
+    AdapterResponse, CacheScope, CacheStoreError, ConfirmationPolicy, CoordinationStateReservation,
+    CoordinationStateStore, CoordinationStateValue, ExecutionConfig, ExecutionMode, HttpMethod,
+    Operation, OperationId, OperationSafetyClass, OperationSafetyPolicy, OperationSecurityLevel,
     OperationStatus, Protocol, ProtocolAdapter, ProtocolAdapterError, RestTarget, Target,
     ToolDescription,
 };
@@ -146,6 +147,111 @@ async fn confirmation_token_allows_only_one_concurrent_execution() {
     );
 }
 
+#[tokio::test]
+async fn unavailable_confirmation_store_is_preserved_as_a_typed_error() {
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(CountingAdapter {
+            call_count: Arc::new(AtomicUsize::new(0)),
+        }))
+        .with_coordination_store(Arc::new(UnavailableCoordinationStore))
+        .build();
+    let operation: crank_runtime::RuntimeOperation = destructive_delete_operation().into();
+    let context = RuntimeRequestContext::from_request_id("req_confirm_unavailable")
+        .with_response_cache_scope("workspace_1", "agent_1");
+
+    let issue_error = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "order_id": "ord_123" }),
+            Some(&context),
+        )
+        .await
+        .expect_err("unavailable store must prevent issuing a token");
+    assert!(matches!(
+        issue_error,
+        RuntimeError::ConfirmationStoreUnavailable { ref operation_id }
+            if operation_id == "op_delete_order"
+    ));
+
+    let consume_error = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "order_id": "ord_123" }),
+            Some(&context.with_confirmation_token("ct_unavailable")),
+        )
+        .await
+        .expect_err("unavailable store must not look like an invalid token");
+    assert!(matches!(
+        consume_error,
+        RuntimeError::ConfirmationStoreUnavailable { ref operation_id }
+            if operation_id == "op_delete_order"
+    ));
+}
+
+struct UnavailableCoordinationStore;
+
+impl UnavailableCoordinationStore {
+    fn error() -> CacheStoreError {
+        CacheStoreError::Unavailable {
+            message: "test backend unavailable".to_owned(),
+        }
+    }
+}
+
+#[async_trait]
+impl CoordinationStateStore for UnavailableCoordinationStore {
+    async fn get_value(
+        &self,
+        _scope: CacheScope,
+        _key: &str,
+    ) -> Result<Option<CoordinationStateValue>, CacheStoreError> {
+        Err(Self::error())
+    }
+
+    async fn put_value(
+        &self,
+        _scope: CacheScope,
+        _key: &str,
+        _value: CoordinationStateValue,
+        _ttl: std::time::Duration,
+    ) -> Result<(), CacheStoreError> {
+        Err(Self::error())
+    }
+
+    async fn delete_value(&self, _scope: CacheScope, _key: &str) -> Result<(), CacheStoreError> {
+        Err(Self::error())
+    }
+
+    async fn take_value(
+        &self,
+        _scope: CacheScope,
+        _key: &str,
+    ) -> Result<Option<CoordinationStateValue>, CacheStoreError> {
+        Err(Self::error())
+    }
+
+    async fn reserve_value(
+        &self,
+        _scope: CacheScope,
+        _key: &str,
+        _value: CoordinationStateValue,
+        _ttl: std::time::Duration,
+    ) -> Result<CoordinationStateReservation, CacheStoreError> {
+        Err(Self::error())
+    }
+
+    async fn compare_and_set_value(
+        &self,
+        _scope: CacheScope,
+        _key: &str,
+        _expected: &CoordinationStateValue,
+        _value: CoordinationStateValue,
+        _ttl: std::time::Duration,
+    ) -> Result<bool, CacheStoreError> {
+        Err(Self::error())
+    }
+}
+
 struct CountingAdapter {
     call_count: Arc<AtomicUsize>,
 }

crates/crank-runtime/tests/product_metrics.rs

@@ -0,0 +1,414 @@
+use std::{
+    collections::BTreeMap,
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+};
+
+use async_trait::async_trait;
+use crank_core::{
+    AdapterResponse, ConfirmationPolicy, ExecutionConfig, ExecutionMode, HttpMethod,
+    IdempotencyMode, IdempotencyPolicy, Operation, OperationId, OperationSafetyClass,
+    OperationSafetyPolicy, OperationSecurityLevel, OperationStatus, Protocol, ProtocolAdapter,
+    ProtocolAdapterError, ResponseCachePolicy, RestTarget,
+    RuntimeRequestContext as ProtocolRequestContext, Target, ToolDescription,
+};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_runtime::{
+    InMemoryCoordinationStateStore, InMemoryResponseCacheStore, RuntimeError,
+    RuntimeExecutorBuilder, RuntimeRequestContext,
+};
+use crank_schema::{Schema, SchemaKind};
+use metrics_util::debugging::DebuggingRecorder;
+use serde_json::json;
+use time::OffsetDateTime;
+
+#[tokio::test]
+async fn real_runtime_paths_emit_cache_idempotency_and_confirmation_outcomes() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    recorder
+        .install()
+        .expect("isolated integration test recorder");
+
+    exercise_response_cache().await;
+    exercise_idempotency().await;
+    exercise_cancelled_idempotency().await;
+    exercise_confirmation().await;
+
+    let snapshot = snapshotter.snapshot().into_vec();
+    for outcome in ["miss", "stored", "hit"] {
+        assert!(has_series(
+            &snapshot,
+            "crank_runtime_cache_total",
+            "outcome",
+            outcome
+        ));
+    }
+    for outcome in [
+        "execute",
+        "completed",
+        "replay",
+        "conflict",
+        "outcome_unknown",
+    ] {
+        assert!(has_series(
+            &snapshot,
+            "crank_idempotency_total",
+            "outcome",
+            outcome
+        ));
+    }
+    for outcome in ["required", "approved", "invalid_token"] {
+        assert!(
+            has_series(&snapshot, "crank_confirmation_total", "outcome", outcome),
+            "missing confirmation outcome {outcome}: {snapshot:?}"
+        );
+    }
+}
+
+async fn exercise_cancelled_idempotency() {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(BlockingAdapter {
+            calls: Arc::clone(&calls),
+        }))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let operation: crank_runtime::RuntimeOperation = operation(
+        "cancelled_idempotent_write",
+        HttpMethod::Post,
+        None,
+        Some(IdempotencyPolicy {
+            mode: IdempotencyMode::Required,
+            ttl_ms: 60_000,
+            input_field: Some("key".to_owned()),
+            header_name: Some("Idempotency-Key".to_owned()),
+        }),
+        None,
+    )
+    .into();
+    let context = RuntimeRequestContext::from_request_id("req_cancelled_idempotency")
+        .with_response_cache_scope("workspace_cancelled", "agent_cancelled");
+    let input = json!({"key": "cancelled-key"});
+
+    let running_executor = executor.clone();
+    let running_operation = operation.clone();
+    let running_context = context.clone();
+    let running_input = input.clone();
+    let task = tokio::spawn(async move {
+        running_executor
+            .execute_with_context(&running_operation, &running_input, Some(&running_context))
+            .await
+    });
+    wait_for_calls(&calls, 1).await;
+    task.abort();
+    let _ = task.await;
+
+    let retry = executor
+        .execute_with_context(&operation, &input, Some(&context))
+        .await
+        .expect_err("cancelled external outcome must not be retried automatically");
+    assert!(matches!(
+        retry,
+        RuntimeError::IdempotencyOutcomeUnknown { .. }
+    ));
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+}
+
+async fn exercise_response_cache() {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let executor = executor(Arc::clone(&calls));
+    let operation: crank_runtime::RuntimeOperation = operation(
+        "cached_lookup",
+        HttpMethod::Get,
+        Some(ResponseCachePolicy { ttl_ms: 60_000 }),
+        None,
+        None,
+    )
+    .into();
+    let context = RuntimeRequestContext::from_request_id("req_cache")
+        .with_response_cache_scope("workspace_cache", "agent_cache");
+
+    executor
+        .execute_with_context(&operation, &json!({"key": "cache-key"}), Some(&context))
+        .await
+        .unwrap();
+    executor
+        .execute_with_context(&operation, &json!({"key": "cache-key"}), Some(&context))
+        .await
+        .unwrap();
+
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+}
+
+async fn exercise_idempotency() {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let executor = executor(Arc::clone(&calls));
+    let operation: crank_runtime::RuntimeOperation = operation(
+        "idempotent_write",
+        HttpMethod::Post,
+        None,
+        Some(IdempotencyPolicy {
+            mode: IdempotencyMode::Required,
+            ttl_ms: 60_000,
+            input_field: Some("key".to_owned()),
+            header_name: Some("Idempotency-Key".to_owned()),
+        }),
+        None,
+    )
+    .into();
+    let context = RuntimeRequestContext::from_request_id("req_idempotency")
+        .with_response_cache_scope("workspace_idempotency", "agent_idempotency");
+
+    let first_input = json!({"key": "stable-key", "variant": "first"});
+    executor
+        .execute_with_context(&operation, &first_input, Some(&context))
+        .await
+        .unwrap();
+    executor
+        .execute_with_context(&operation, &first_input, Some(&context))
+        .await
+        .unwrap();
+    let conflict = executor
+        .execute_with_context(
+            &operation,
+            &json!({"key": "stable-key", "variant": "different"}),
+            Some(&context),
+        )
+        .await
+        .expect_err("same idempotency key with a different input must conflict");
+
+    assert!(matches!(conflict, RuntimeError::IdempotencyConflict { .. }));
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+}
+
+async fn exercise_confirmation() {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let executor = executor(Arc::clone(&calls));
+    let operation: crank_runtime::RuntimeOperation = operation(
+        "destructive_write",
+        HttpMethod::Delete,
+        None,
+        None,
+        Some(OperationSafetyPolicy {
+            class: OperationSafetyClass::Destructive,
+            confirmation: Some(ConfirmationPolicy { ttl_ms: 60_000 }),
+        }),
+    )
+    .into();
+    let context = RuntimeRequestContext::from_request_id("req_confirmation")
+        .with_response_cache_scope("workspace_confirmation", "agent_confirmation");
+    let input = json!({"key": "delete-key"});
+
+    let required = executor
+        .execute_with_context(&operation, &input, Some(&context))
+        .await
+        .expect_err("destructive operation must require confirmation");
+    let RuntimeError::ConfirmationRequired {
+        confirmation_token, ..
+    } = required
+    else {
+        panic!("expected confirmation token");
+    };
+    let confirmed = context
+        .clone()
+        .with_confirmation_token(confirmation_token.clone());
+    executor
+        .execute_with_context(&operation, &input, Some(&confirmed))
+        .await
+        .unwrap();
+    let invalid = executor
+        .execute_with_context(&operation, &input, Some(&confirmed))
+        .await
+        .expect_err("confirmation token must be single-use");
+
+    assert!(matches!(
+        invalid,
+        RuntimeError::InvalidConfirmationToken { .. }
+    ));
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+}
+
+fn executor(calls: Arc<AtomicUsize>) -> crank_runtime::RuntimeExecutor {
+    RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(CountingAdapter { calls }))
+        .with_response_cache(Arc::new(InMemoryResponseCacheStore::default()))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build()
+}
+
+struct CountingAdapter {
+    calls: Arc<AtomicUsize>,
+}
+
+struct BlockingAdapter {
+    calls: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl ProtocolAdapter for BlockingAdapter {
+    fn protocol(&self) -> Protocol {
+        Protocol::Rest
+    }
+
+    fn supports_mode(&self, mode: ExecutionMode) -> bool {
+        mode == ExecutionMode::Unary
+    }
+
+    async fn invoke_unary(
+        &self,
+        _target: &Target,
+        _prepared: &crank_core::PreparedRequest,
+        _context: &ProtocolRequestContext,
+    ) -> Result<AdapterResponse, ProtocolAdapterError> {
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        std::future::pending().await
+    }
+}
+
+#[async_trait]
+impl ProtocolAdapter for CountingAdapter {
+    fn protocol(&self) -> Protocol {
+        Protocol::Rest
+    }
+
+    fn supports_mode(&self, mode: ExecutionMode) -> bool {
+        mode == ExecutionMode::Unary
+    }
+
+    async fn invoke_unary(
+        &self,
+        _target: &Target,
+        _prepared: &crank_core::PreparedRequest,
+        _context: &ProtocolRequestContext,
+    ) -> Result<AdapterResponse, ProtocolAdapterError> {
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        Ok(AdapterResponse {
+            status_code: 200,
+            headers: BTreeMap::new(),
+            body: json!({"result": "ok"}),
+            data: json!({"result": "ok"}),
+        })
+    }
+}
+
+fn operation(
+    name: &str,
+    method: HttpMethod,
+    response_cache: Option<ResponseCachePolicy>,
+    idempotency: Option<IdempotencyPolicy>,
+    safety: Option<OperationSafetyPolicy>,
+) -> Operation<Schema, MappingSet> {
+    Operation {
+        id: OperationId::new(format!("op_{name}")),
+        name: name.to_owned(),
+        display_name: name.to_owned(),
+        category: "metrics".to_owned(),
+        protocol: Protocol::Rest,
+        security_level: OperationSecurityLevel::Standard,
+        status: OperationStatus::Published,
+        version: 1,
+        target: Target::Rest(RestTarget {
+            base_url: "https://metrics.example.invalid".to_owned(),
+            method,
+            path_template: "/resource".to_owned(),
+            static_headers: BTreeMap::new(),
+        }),
+        input_schema: object_schema("key"),
+        output_schema: object_schema("result"),
+        input_mapping: MappingSet { rules: Vec::new() },
+        output_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.response.body.result".to_owned(),
+                target: "$.output.result".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        execution_config: ExecutionConfig {
+            timeout_ms: 1_000,
+            retry_policy: None,
+            response_cache,
+            idempotency,
+            safety,
+            approval_policy: None,
+            auth_profile_ref: None,
+            headers: BTreeMap::new(),
+        },
+        tool_description: ToolDescription {
+            title: name.to_owned(),
+            description: "Exercises a real runtime metrics path.".to_owned(),
+            tags: Vec::new(),
+            examples: Vec::new(),
+        },
+        samples: None,
+        generated_draft: None,
+        config_export: None,
+        wizard_state: None,
+        created_at: OffsetDateTime::UNIX_EPOCH,
+        updated_at: OffsetDateTime::UNIX_EPOCH,
+        published_at: None,
+    }
+}
+
+fn object_schema(field: &str) -> Schema {
+    Schema {
+        kind: SchemaKind::Object,
+        description: None,
+        required: true,
+        nullable: false,
+        default_value: None,
+        fields: BTreeMap::from([(
+            field.to_owned(),
+            Schema {
+                kind: SchemaKind::String,
+                description: None,
+                required: true,
+                nullable: false,
+                default_value: None,
+                fields: BTreeMap::new(),
+                items: None,
+                enum_values: Vec::new(),
+                variants: Vec::new(),
+            },
+        )]),
+        items: None,
+        enum_values: Vec::new(),
+        variants: Vec::new(),
+    }
+}
+
+fn has_series(
+    snapshot: &[(
+        metrics_util::CompositeKey,
+        Option<metrics::Unit>,
+        Option<metrics::SharedString>,
+        metrics_util::debugging::DebugValue,
+    )],
+    metric: &str,
+    label_name: &str,
+    label_value: &str,
+) -> bool {
+    snapshot.iter().any(|(key, _, _, _)| {
+        key.key().name() == metric
+            && key
+                .key()
+                .labels()
+                .any(|label| label.key() == label_name && label.value() == label_value)
+    })
+}
+
+async fn wait_for_calls(calls: &AtomicUsize, expected: usize) {
+    for _ in 0..1_000 {
+        if calls.load(Ordering::SeqCst) >= expected {
+            return;
+        }
+        tokio::task::yield_now().await;
+    }
+    panic!("adapter did not receive {expected} call(s)");
+}

docs/observability.md

@@ -99,6 +99,11 @@ CRANK_SENTRY_DSN=https://public-key@errors.example.com/1
 - доступные `request_id` и `trace_id`;
 - статическое сообщение без исходного текста ошибки.
 
+События группируются по паре `service` и закрытой категории, поэтому одинаковая
+ошибка `admin-api` и `mcp-server` не объединяется в один инцидент. Если
+обязательные поля идентичности не помещаются в заданный предел события, запуск
+останавливается безопасной типизированной ошибкой до приёма запросов.
+
 До отправки удаляются request, user, breadcrumbs, URL, query, cookie,
 authorization, payload, произвольные contexts и extra, а также исходный текст
 panic или ошибки. Performance tracing, журналы, показатели и отслеживание
@@ -165,11 +170,14 @@ scrape_configs:
 
 - `crank_http_requests_total`, `crank_http_request_duration_seconds`,
   `crank_http_inflight`;
-- `crank_mcp_requests_total`, `crank_mcp_active_sessions`;
+- `crank_mcp_requests_total`, `crank_mcp_active_sessions`,
+  `crank_mcp_active_streams`;
 - `crank_tool_invocations_total`, `crank_tool_invocation_duration_seconds`;
 - `crank_upstream_requests_total`,
   `crank_upstream_request_duration_seconds`;
-- `crank_runtime_inflight`, `crank_runtime_limit_rejections_total`;
+- `crank_runtime_inflight`, `crank_runtime_limit_rejections_total`,
+  `crank_runtime_cache_total`, `crank_idempotency_total`,
+  `crank_confirmation_total`;
 - `crank_db_pool_connections`;
 - `crank_catalog_tools`, `crank_catalog_estimated_context_tokens`,
   `crank_catalog_warnings`;
@@ -178,11 +186,25 @@ scrape_configs:
 
 Recorder добавляет к рядам проверенные статические labels `service`,
 `version`, `environment`. Остальные labels имеют закрытый набор значений.
+Имена, типы и допустимые значения labels определены в независимом нижнем
+crate `crank-metrics`. Product-код вызывает только его типизированный фасад;
+прямая production-зависимость от `metrics` вне `crank-metrics` и
+`crank-observability` запрещена архитектурной проверкой Cargo metadata.
 Запрещено использовать workspace, идентификаторы агента, операции или
 запроса, фактический URL, текст ошибки, payload и пользовательский текст.
 HTTP route берётся только из шаблона Axum; неизвестные маршруты и методы
 сворачиваются в `unmatched` и `OTHER`.
 
+`crank_mcp_active_sessions` отражает число ещё не истёкших транспортных
+сессий в общем хранилище, а `crank_mcp_active_streams` — число открытых в
+этом экземпляре SSE-потоков. Прикладные ошибки JSON-RPC и
+`tools/call.result.isError=true` учитываются отдельно от успешного HTTP 200.
+Счётчики cache, idempotency и confirmation описывают переходы
+соответствующих стадий: hit/miss и ошибки хранилища, execute/replay/conflict
+и required/approved/invalid token. Отменённый future runtime или upstream
+завершает соответствующее измерение закрытым исходом `aborted`, поэтому
+фактически начатый вызов не исчезает из показателей.
+
 Buckets длительности фиксированы кодом:
 `0.005`, `0.01`, `0.025`, `0.05`, `0.1`, `0.25`, `0.5`, `1`, `2.5`, `5`,
 `10`, `30`, `60` секунд. Это пока техническая шкала, а не SLO. Фактические

docs/runtime-config.md

@@ -151,6 +151,11 @@ CRANK_LOG_LEVEL=info
 Внешний сборщик журналов не обязателен. Его отсутствие не влияет на `/health`
 и `/ready`.
 
+`CRANK_INVOCATION_LOG_RETENTION_DAYS` задаёт срок хранения подробной истории
+вызовов в днях. Допустимый диапазон: от `1` до `36500`, значение по умолчанию:
+`30`. Значение вне диапазона останавливает запуск до создания фоновой задачи
+очистки.
+
 ## Критические ошибки
 
 - `CRANK_SENTRY_DSN` — DSN внешнего Sentry-совместимого приёмника.

scripts/check-rust-boundaries.py

@@ -49,6 +49,8 @@ def package_category(name: str, manifest_path: Path, workspace_root: Path) -> st
         return "app"
     if name == "crank-core":
         return "core"
+    if name == "crank-metrics":
+        return "metrics"
     if name == "crank-observability":
         return "observability"
     if name == "crank-registry":
@@ -92,6 +94,23 @@ def dependency_package_ids(raw_package: dict[str, Any], packages_by_name: dict[s
     return dependency_ids
 
 
+def direct_dependency_violations(raw_package: dict[str, Any]) -> list[Violation]:
+    package_name = raw_package["name"]
+    if package_name in {"crank-metrics", "crank-observability"}:
+        return []
+
+    return [
+        Violation(
+            source=package_name,
+            dependency="metrics",
+            reason="production crates must record metrics through crank-metrics",
+        )
+        for dependency in raw_package.get("dependencies", [])
+        if dependency["name"] == "metrics"
+        and dependency.get("kind") != "dev"
+    ]
+
+
 def boundary_reason(source: Package, dependency: Package) -> str | None:
     if source.category == "app":
         if dependency.category == "app":
@@ -101,7 +120,10 @@ def boundary_reason(source: Package, dependency: Package) -> str | None:
     if dependency.category == "app":
         return "workspace crates must not depend on apps"
 
-    if source.category == "observability":
+    if source.category == "metrics":
+        return "crank-metrics must not depend on other workspace crates"
+
+    if source.category == "observability" and dependency.category != "metrics":
         return "crank-observability must not depend on other workspace crates"
 
     if (
@@ -133,6 +155,7 @@ def find_violations(metadata: dict[str, Any]) -> list[Violation]:
 
     for source in sorted(packages.values(), key=lambda package: package.name):
         raw_package = raw_packages_by_id[source.id]
+        violations.extend(direct_dependency_violations(raw_package))
         for dependency_id in dependency_package_ids(raw_package, packages_by_name):
             dependency = packages[dependency_id]
             reason = boundary_reason(source, dependency)

scripts/check-rust-module-boundaries.sh

@@ -47,6 +47,13 @@ check_no_match \
   '^\s*use\s+(axum|sqlx)(::|[;\{])' \
   "$ROOT_DIR/crates/crank-runtime/src"
 
+check_no_match \
+  "product modules must record metrics only through crank-metrics" \
+  '(^|[^[:alnum:]_])(::)?metrics::(counter|gauge|histogram)!' \
+  "$ROOT_DIR/apps" \
+  "$ROOT_DIR/crates" \
+  --glob '!**/crank-metrics/**'
+
 if (( status != 0 )); then
   cat >&2 <<'EOF'
 
@@ -58,6 +65,7 @@ Rules:
 - core remains framework/storage agnostic;
 - registry remains storage-only and HTTP-client agnostic;
 - runtime remains execution-only and storage/framework agnostic.
+- names and labels of metrics remain inside the typed crank-metrics contract.
 
 EOF
 fi

tests/unit/test_check_rust_boundaries.py

@@ -27,6 +27,10 @@ def package(root: Path, name: str, rel_dir: str, dependencies: list[str] | None
     }
 
 
+def dependency(name: str, kind: str | None = None) -> dict:
+    return {"name": name, "kind": kind}
+
+
 def metadata(packages: list[dict], root: Path | None = None) -> dict:
     root = root or Path("/tmp/crank")
     return {
@@ -124,6 +128,44 @@ class RustBoundaryCheckTests(unittest.TestCase):
             self.assertEqual(violations[0].source, "crank-observability")
             self.assertEqual(violations[0].dependency, dependency)
 
+    def test_allows_observability_and_product_crates_to_depend_on_metrics_contract(self) -> None:
+        packages = [
+            package(
+                self.root,
+                "crank-observability",
+                "crates/crank-observability",
+                ["crank-metrics"],
+            ),
+            package(
+                self.root,
+                "crank-runtime",
+                "crates/crank-runtime",
+                ["crank-metrics"],
+            ),
+            package(self.root, "crank-metrics", "crates/crank-metrics"),
+        ]
+
+        violations = self.checker.find_violations(metadata(packages, self.root))
+
+        self.assertEqual(violations, [])
+
+    def test_rejects_metrics_contract_dependency_on_workspace_crates(self) -> None:
+        packages = [
+            package(
+                self.root,
+                "crank-metrics",
+                "crates/crank-metrics",
+                ["crank-core"],
+            ),
+            package(self.root, "crank-core", "crates/crank-core"),
+        ]
+
+        violations = self.checker.find_violations(metadata(packages, self.root))
+
+        self.assertEqual(len(violations), 1)
+        self.assertEqual(violations[0].source, "crank-metrics")
+        self.assertEqual(violations[0].dependency, "crank-core")
+
     def test_rejects_domain_and_runtime_dependencies_on_observability(self) -> None:
         for source in ("crank-core", "crank-registry", "crank-runtime"):
             packages = [
@@ -146,6 +188,23 @@ class RustBoundaryCheckTests(unittest.TestCase):
             self.assertEqual(violations[0].source, source)
             self.assertEqual(violations[0].dependency, "crank-observability")
 
+    def test_rejects_direct_production_metrics_dependency_outside_contract(self) -> None:
+        app = package(self.root, "admin-api", "apps/admin-api")
+        app["dependencies"] = [dependency("metrics")]
+
+        violations = self.checker.find_violations(metadata([app], self.root))
+
+        self.assertEqual(len(violations), 1)
+        self.assertEqual(violations[0].dependency, "metrics")
+
+    def test_allows_metrics_as_test_only_dependency(self) -> None:
+        app = package(self.root, "admin-api", "apps/admin-api")
+        app["dependencies"] = [dependency("metrics", "dev")]
+
+        violations = self.checker.find_violations(metadata([app], self.root))
+
+        self.assertEqual(violations, [])
+
 
 if __name__ == "__main__":
     unittest.main()