наблюдаемость: завершить базовый контур Community

Добавить структурированные журналы, метрики, трассировку и безопасный канал критических ошибок. Усилить границы рантайма, тесты, проверку зависимостей и сценарии развёртывания.

crates/crank-runtime/Cargo.toml

@@ -3,6 +3,7 @@ name = "crank-runtime"
 edition.workspace = true
 license.workspace = true
 rust-version.workspace = true
+publish.workspace = true
 version.workspace = true
 
 [lib]
@@ -19,19 +20,22 @@ crank-adapter-rest = { path = "../crank-adapter-rest" }
 crank-core = { path = "../crank-core" }
 crank-mapping = { path = "../crank-mapping" }
 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
 sha2.workspace = true
 thiserror.workspace = true
 time.workspace = true
-tokio = { workspace = true, features = ["sync"] }
+tokio = { workspace = true, features = ["sync", "time"] }
 tracing.workspace = true
 uuid.workspace = true
 
 [dev-dependencies]
 axum.workspace = true
 futures-util = "0.3"
+testcontainers.workspace = true
 time.workspace = true
 tracing-subscriber.workspace = true

crates/crank-runtime/src/cache.rs

@@ -8,9 +8,9 @@ use std::{
 
 use async_trait::async_trait;
 use crank_core::{
-    CacheBackend, CacheScope, CacheStoreError, CachedResponse, CoordinationStateStore,
-    CoordinationStateValue, RateLimitBucketState, RateLimitStateStore, ReplayGuardStatus,
-    ReplayGuardStore, ResponseCacheStore,
+    CacheBackend, CacheScope, CacheStoreError, CachedResponse, CoordinationStateReservation,
+    CoordinationStateStore, CoordinationStateValue, RateLimitBucketState, RateLimitDecision,
+    RateLimitStateStore, ReplayGuardStatus, ReplayGuardStore, ResponseCacheStore,
 };
 use redis::{Client, aio::ConnectionManager};
 use serde::de::DeserializeOwned;
@@ -307,6 +307,43 @@ impl RateLimitStateStore for InMemoryRateLimitStateStore {
         entries.remove(key);
         Ok(())
     }
+
+    async fn consume_token(
+        &self,
+        key: &str,
+        burst_tokens_micros: u64,
+        refill_per_second_micros: u64,
+        now_unix_ms: i64,
+        ttl: Duration,
+    ) -> Result<RateLimitDecision, CacheStoreError> {
+        validate_key(key)?;
+        validate_rate_limit_parameters(burst_tokens_micros, refill_per_second_micros)?;
+        let expires_at = expiry_from_ttl(ttl)?;
+        let now = Instant::now();
+        let mut entries = self.entries.write().await;
+        retain_unexpired(&mut entries, now);
+        let state = entries
+            .get(key)
+            .map(|entry| entry.value)
+            .unwrap_or(RateLimitBucketState {
+                tokens_micros: burst_tokens_micros,
+                last_refill_unix_ms: now_unix_ms,
+            });
+        let (state, decision) = consume_bucket_token(
+            state,
+            burst_tokens_micros,
+            refill_per_second_micros,
+            now_unix_ms,
+        );
+        entries.insert(
+            key.to_owned(),
+            ExpiringValue {
+                value: state,
+                expires_at,
+            },
+        );
+        Ok(decision)
+    }
 }
 
 #[async_trait]
@@ -373,6 +410,64 @@ impl CoordinationStateStore for InMemoryCoordinationStateStore {
         entries.remove(&storage_key);
         Ok(())
     }
+
+    async fn take_value(
+        &self,
+        scope: CacheScope,
+        key: &str,
+    ) -> Result<Option<CoordinationStateValue>, CacheStoreError> {
+        validate_key(key)?;
+        let storage_key = scoped_key(scope, key);
+        let now = Instant::now();
+        let mut entries = self.entries.write().await;
+        retain_unexpired(&mut entries, now);
+        Ok(entries.remove(&storage_key).map(|entry| entry.value))
+    }
+
+    async fn reserve_value(
+        &self,
+        scope: CacheScope,
+        key: &str,
+        value: CoordinationStateValue,
+        ttl: Duration,
+    ) -> Result<CoordinationStateReservation, CacheStoreError> {
+        validate_key(key)?;
+        let expires_at = expiry_from_ttl(ttl)?;
+        let storage_key = scoped_key(scope, key);
+        let now = Instant::now();
+        let mut entries = self.entries.write().await;
+        retain_unexpired(&mut entries, now);
+        if let Some(existing) = entries.get(&storage_key) {
+            return Ok(CoordinationStateReservation::Existing(
+                existing.value.clone(),
+            ));
+        }
+        entries.insert(storage_key, ExpiringValue { value, expires_at });
+        Ok(CoordinationStateReservation::Reserved)
+    }
+
+    async fn compare_and_set_value(
+        &self,
+        scope: CacheScope,
+        key: &str,
+        expected: &CoordinationStateValue,
+        value: CoordinationStateValue,
+        ttl: Duration,
+    ) -> Result<bool, CacheStoreError> {
+        validate_key(key)?;
+        let expires_at = expiry_from_ttl(ttl)?;
+        let storage_key = scoped_key(scope, key);
+        let now = Instant::now();
+        let mut entries = self.entries.write().await;
+        retain_unexpired(&mut entries, now);
+        let matches = entries
+            .get(&storage_key)
+            .is_some_and(|entry| entry.value == *expected);
+        if matches {
+            entries.insert(storage_key, ExpiringValue { value, expires_at });
+        }
+        Ok(matches)
+    }
 }
 
 #[async_trait]
@@ -413,6 +508,65 @@ impl RateLimitStateStore for RedisCacheStore {
     async fn delete_bucket(&self, key: &str) -> Result<(), CacheStoreError> {
         self.delete_kind_key("rate_limit", key).await
     }
+
+    async fn consume_token(
+        &self,
+        key: &str,
+        burst_tokens_micros: u64,
+        refill_per_second_micros: u64,
+        now_unix_ms: i64,
+        ttl: Duration,
+    ) -> Result<RateLimitDecision, CacheStoreError> {
+        validate_key(key)?;
+        validate_rate_limit_parameters(burst_tokens_micros, refill_per_second_micros)?;
+        let storage_key = self.prefixed_key("rate_limit", key);
+        let ttl_ms = self.ttl_ms(ttl)?;
+        let mut connection = self.connection_manager.clone();
+        let script = r#"
+local current = redis.call('GET', KEYS[1])
+local tokens = tonumber(ARGV[1])
+local last_refill = tonumber(ARGV[3])
+if current then
+  local decoded = cjson.decode(current)
+  tokens = tonumber(decoded.tokens_micros)
+  last_refill = tonumber(decoded.last_refill_unix_ms)
+end
+local effective_now = math.max(tonumber(ARGV[3]), last_refill)
+local elapsed = effective_now - last_refill
+local replenished = math.floor(elapsed * tonumber(ARGV[2]) / 1000)
+tokens = math.min(tonumber(ARGV[1]), tokens + replenished)
+local allowed = 0
+local retry_after = 0
+if tokens >= 1000000 then
+  tokens = tokens - 1000000
+  allowed = 1
+else
+  local missing = 1000000 - tokens
+  retry_after = math.max(1, math.ceil(missing * 1000 / tonumber(ARGV[2])))
+end
+redis.call('PSETEX', KEYS[1], ARGV[4], cjson.encode({
+  tokens_micros = tokens,
+  last_refill_unix_ms = effective_now
+}))
+return {allowed, retry_after}
+"#;
+        let (allowed, retry_after_ms): (u8, u64) = redis::cmd("EVAL")
+            .arg(script)
+            .arg(1)
+            .arg(storage_key)
+            .arg(burst_tokens_micros)
+            .arg(refill_per_second_micros)
+            .arg(now_unix_ms)
+            .arg(ttl_ms)
+            .query_async(&mut connection)
+            .await
+            .map_err(|source| self.unavailable(source))?;
+        Ok(if allowed == 1 {
+            RateLimitDecision::Allowed
+        } else {
+            RateLimitDecision::Rejected { retry_after_ms }
+        })
+    }
 }
 
 #[async_trait]
@@ -472,6 +626,126 @@ impl CoordinationStateStore for RedisCacheStore {
         self.delete_kind_key(Self::coordination_kind(scope), key)
             .await
     }
+
+    async fn take_value(
+        &self,
+        scope: CacheScope,
+        key: &str,
+    ) -> Result<Option<CoordinationStateValue>, CacheStoreError> {
+        validate_key(key)?;
+        let storage_key = self.prefixed_key(Self::coordination_kind(scope), key);
+        let mut connection = self.connection_manager.clone();
+        let encoded: Option<Vec<u8>> = redis::cmd("EVAL")
+            .arg("local value = redis.call('GET', KEYS[1]); if value then redis.call('DEL', KEYS[1]); end; return value")
+            .arg(1)
+            .arg(storage_key)
+            .query_async(&mut connection)
+            .await
+            .map_err(|source| self.unavailable(source))?;
+        encoded
+            .map(|bytes| self.deserialize_value(&bytes))
+            .transpose()
+    }
+
+    async fn reserve_value(
+        &self,
+        scope: CacheScope,
+        key: &str,
+        value: CoordinationStateValue,
+        ttl: Duration,
+    ) -> Result<CoordinationStateReservation, CacheStoreError> {
+        validate_key(key)?;
+        let storage_key = self.prefixed_key(Self::coordination_kind(scope), key);
+        let encoded = self.serialize_value(&value)?;
+        let ttl_ms = self.ttl_ms(ttl)?;
+        let mut connection = self.connection_manager.clone();
+        let (reserved, existing): (u8, Vec<u8>) = redis::cmd("EVAL")
+            .arg("local current = redis.call('GET', KEYS[1]); if current then return {0, current}; end; redis.call('PSETEX', KEYS[1], ARGV[2], ARGV[1]); return {1, ''}")
+            .arg(1)
+            .arg(storage_key)
+            .arg(encoded)
+            .arg(ttl_ms)
+            .query_async(&mut connection)
+            .await
+            .map_err(|source| self.unavailable(source))?;
+        if reserved == 1 {
+            Ok(CoordinationStateReservation::Reserved)
+        } else {
+            Ok(CoordinationStateReservation::Existing(
+                self.deserialize_value(&existing)?,
+            ))
+        }
+    }
+
+    async fn compare_and_set_value(
+        &self,
+        scope: CacheScope,
+        key: &str,
+        expected: &CoordinationStateValue,
+        value: CoordinationStateValue,
+        ttl: Duration,
+    ) -> Result<bool, CacheStoreError> {
+        validate_key(key)?;
+        let storage_key = self.prefixed_key(Self::coordination_kind(scope), key);
+        let expected = self.serialize_value(expected)?;
+        let value = self.serialize_value(&value)?;
+        let ttl_ms = self.ttl_ms(ttl)?;
+        let mut connection = self.connection_manager.clone();
+        let replaced: u8 = redis::cmd("EVAL")
+            .arg("local current = redis.call('GET', KEYS[1]); if current ~= ARGV[1] then return 0; end; redis.call('PSETEX', KEYS[1], ARGV[3], ARGV[2]); return 1")
+            .arg(1)
+            .arg(storage_key)
+            .arg(expected)
+            .arg(value)
+            .arg(ttl_ms)
+            .query_async(&mut connection)
+            .await
+            .map_err(|source| self.unavailable(source))?;
+        Ok(replaced == 1)
+    }
+}
+
+fn consume_bucket_token(
+    mut state: RateLimitBucketState,
+    burst_tokens_micros: u64,
+    refill_per_second_micros: u64,
+    now_unix_ms: i64,
+) -> (RateLimitBucketState, RateLimitDecision) {
+    let effective_now = now_unix_ms.max(state.last_refill_unix_ms);
+    let elapsed_ms =
+        u64::try_from(effective_now.saturating_sub(state.last_refill_unix_ms)).unwrap_or(u64::MAX);
+    let replenished =
+        u128::from(elapsed_ms).saturating_mul(u128::from(refill_per_second_micros)) / 1000;
+    state.tokens_micros = u128::from(state.tokens_micros)
+        .saturating_add(replenished)
+        .min(u128::from(burst_tokens_micros))
+        .try_into()
+        .unwrap_or(burst_tokens_micros);
+    state.last_refill_unix_ms = effective_now;
+    if state.tokens_micros >= 1_000_000 {
+        state.tokens_micros -= 1_000_000;
+        return (state, RateLimitDecision::Allowed);
+    }
+    let missing = 1_000_000_u64.saturating_sub(state.tokens_micros);
+    let retry_after_ms = u128::from(missing)
+        .saturating_mul(1000)
+        .div_ceil(u128::from(refill_per_second_micros))
+        .max(1)
+        .try_into()
+        .unwrap_or(u64::MAX);
+    (state, RateLimitDecision::Rejected { retry_after_ms })
+}
+
+fn validate_rate_limit_parameters(
+    burst_tokens_micros: u64,
+    refill_per_second_micros: u64,
+) -> Result<(), CacheStoreError> {
+    if burst_tokens_micros < 1_000_000 || refill_per_second_micros == 0 {
+        return Err(CacheStoreError::InvalidKey {
+            message: "rate limit capacity and refill rate must be positive".to_owned(),
+        });
+    }
+    Ok(())
 }
 
 fn validate_key(key: &str) -> Result<(), CacheStoreError> {
@@ -530,7 +804,11 @@ fn parse_optional_string(name: &'static str) -> Result<Option<String>, RuntimeCa
 fn parse_optional_u64(name: &'static str) -> Result<Option<u64>, RuntimeCacheConfigError> {
     match env::var(name) {
         Ok(raw) => {
-            let value = raw
+            let trimmed = raw.trim();
+            if trimmed.is_empty() {
+                return Ok(None);
+            }
+            let value = trimmed
                 .parse::<u64>()
                 .map_err(|source| RuntimeCacheConfigError::InvalidTtl { value: raw, source })?;
             if value == 0 {

crates/crank-runtime/src/confirmation.rs

@@ -50,6 +50,12 @@ pub async fn confirm_operation(
     consume_confirmation_token(store, &scope, provided_token, &input_hash).await
 }
 
+pub(crate) fn is_applicable(operation: &RuntimeOperation) -> bool {
+    effective_safety_policy(operation)
+        .class
+        .requires_confirmation()
+}
+
 fn effective_safety_policy(operation: &RuntimeOperation) -> OperationSafetyPolicy {
     operation
         .execution_config
@@ -137,13 +143,12 @@ async fn consume_confirmation_token(
 ) -> Result<(), RuntimeError> {
     let key = confirmation_cache_key(operation_scope, token);
     let stored = store
-        .get_value(CacheScope::Coordination, &key)
+        .take_value(CacheScope::Coordination, &key)
         .await
         .map_err(|error| RuntimeError::InvalidPreparedRequest {
             field: "confirmation_token".to_owned(),
             reason: error.to_string(),
         })?;
-    let _ = store.delete_value(CacheScope::Coordination, &key).await;
 
     let Some(stored) = stored else {
         return Err(RuntimeError::InvalidConfirmationToken {

crates/crank-runtime/src/error.rs

@@ -38,6 +38,14 @@ pub enum RuntimeError {
     InvalidConfirmationToken { operation_id: String },
     #[error("confirmation store is unavailable for operation {operation_id}")]
     ConfirmationStoreUnavailable { operation_id: String },
+    #[error("idempotency store is unavailable for operation {operation_id}")]
+    IdempotencyStoreUnavailable { operation_id: String },
+    #[error("operation {operation_id} is already executing for this idempotency key")]
+    IdempotencyInProgress { operation_id: String },
+    #[error("idempotency key for operation {operation_id} was reused with different input")]
+    IdempotencyConflict { operation_id: String },
+    #[error("the outcome of operation {operation_id} is unknown; automatic retry is unsafe")]
+    IdempotencyOutcomeUnknown { operation_id: String },
     #[error("auth profile {auth_profile_id} was not found")]
     MissingAuthProfile { auth_profile_id: String },
     #[error("secret {secret_id} was not found")]

crates/crank-runtime/src/executor.rs

@@ -1,14 +1,16 @@
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use crank_core::{
-    AdapterRegistry, CoordinationStateStore, ExecutionMode, InvocationStatus, MeteringEvent,
-    ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter,
+    AdapterRegistry, CoordinationStateStore, ExecutionMode, InvocationSource, InvocationStatus,
+    MeteringEvent, ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter,
 };
+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::debug;
+use tracing::{Instrument, Span, debug};
 
 use crate::{
     AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeLimits, RuntimeOperation,
@@ -176,18 +178,31 @@ impl RuntimeExecutor {
         request: RuntimeExecutionRequest<'_>,
     ) -> Result<Value, RuntimeError> {
         log_runtime_event("unary.execute", request.operation, request.request_context);
-        let _permit = self.acquire_unary_permit(request.operation)?;
         let started_at = Instant::now();
-        let prepared_request = self.prepare_request(request.operation, request.input)?;
-        let prepared_request = apply_resolved_auth(prepared_request, request.resolved_auth);
-        let result = self
-            .execute_prepared(
+        let runtime_span = Stage::RuntimeExecute.span();
+        let result = async {
+            let _permit = self.acquire_unary_permit(request.operation)?;
+            let _inflight = RuntimeInFlightGuard::new();
+            let mapping_span = Stage::RuntimeArgumentsMap.span();
+            let prepared_request =
+                mapping_span.in_scope(|| self.prepare_request(request.operation, request.input));
+            record_runtime_result(&mapping_span, &prepared_request);
+            drop(mapping_span);
+            let prepared_request = prepared_request?;
+            let prepared_request = apply_resolved_auth(prepared_request, request.resolved_auth);
+            self.execute_prepared(
                 request.operation,
                 request.input,
                 prepared_request,
                 request.request_context,
             )
-            .await;
+            .await
+        }
+        .instrument(runtime_span.clone())
+        .await;
+        record_runtime_result(&runtime_span, &result);
+        drop(runtime_span);
+        record_execution_metrics(request.request_context, &result, started_at);
         self.record_metering(
             request.operation,
             request.request_context,
@@ -221,59 +236,132 @@ impl RuntimeExecutor {
         request_context: Option<&RuntimeRequestContext>,
     ) -> Result<Value, RuntimeError> {
         let mut prepared_request = prepared_request;
-        let idempotency_key =
-            crate::idempotency::prepare_idempotency(operation, input, &mut prepared_request)?;
-        crate::confirmation::confirm_operation(
-            self.coordination_store.as_deref(),
+        let idempotency_applicable = crate::idempotency::policy(operation).is_some();
+        let idempotency_key = match crate::idempotency::prepare_idempotency(
             operation,
             input,
-            request_context,
-        )
-        .await?;
-        if let Some(response) = self
-            .load_idempotent_adapter_response(
+            &mut prepared_request,
+        ) {
+            Ok(key) => key,
+            Err(error) if idempotency_applicable => {
+                let span = Stage::RuntimeIdempotency.span();
+                StageOutcome::Error.record(&span);
+                ErrorCategory::Idempotency.record(&span);
+                return Err(error);
+            }
+            Err(error) => return Err(error),
+        };
+
+        if crate::confirmation::is_applicable(operation) {
+            let approval_span = Stage::ApprovalCheck.span();
+            let approval_result = crate::confirmation::confirm_operation(
+                self.coordination_store.as_deref(),
                 operation,
+                input,
+                request_context,
+            )
+            .instrument(approval_span.clone())
+            .await;
+            match &approval_result {
+                Ok(()) => StageOutcome::Success.record(&approval_span),
+                Err(RuntimeError::ConfirmationRequired { .. }) => {
+                    StageOutcome::Required.record(&approval_span);
+                    ErrorCategory::Approval.record(&approval_span);
+                }
+                Err(error) => {
+                    StageOutcome::Error.record(&approval_span);
+                    runtime_error_category(error).record(&approval_span);
+                }
+            }
+            drop(approval_span);
+            approval_result?;
+        }
+
+        let idempotency = if idempotency_applicable {
+            let idempotency_span = Stage::RuntimeIdempotency.span();
+            let result = crate::idempotency::begin(
+                self.coordination_store.as_deref(),
+                operation,
+                input,
                 idempotency_key.as_deref(),
                 request_context,
             )
-            .await
-        {
-            let finalized_output = finalize_output(operation, &response)?;
-            operation.output_schema.validate_shape(&finalized_output)?;
-            return Ok(finalized_output);
+            .instrument(idempotency_span.clone())
+            .await;
+            match &result {
+                Ok(crate::idempotency::IdempotencyAction::Execute(_)) => {
+                    StageOutcome::Execute.record(&idempotency_span);
+                }
+                Ok(crate::idempotency::IdempotencyAction::Replay(_)) => {
+                    StageOutcome::Replay.record(&idempotency_span);
+                }
+                Ok(crate::idempotency::IdempotencyAction::Disabled) => {
+                    StageOutcome::Skipped.record(&idempotency_span);
+                }
+                Err(error) => {
+                    StageOutcome::Error.record(&idempotency_span);
+                    runtime_error_category(error).record(&idempotency_span);
+                }
+            }
+            drop(idempotency_span);
+            result?
+        } else {
+            crate::idempotency::IdempotencyAction::Disabled
+        };
+        if let crate::idempotency::IdempotencyAction::Replay(response) = &idempotency {
+            return transform_response(operation, response);
         }
 
-        let adapter_response = match self
+        let adapter_result = match self
             .load_cached_adapter_response(operation, &prepared_request, request_context)
             .await
         {
-            Some(response) => response,
+            Some(response) => Ok(response),
             None => {
                 let adapter_response = self
                     .execute_adapter(operation, prepared_request.clone(), request_context)
-                    .await?;
-                self.store_cached_adapter_response(
-                    operation,
-                    &prepared_request,
-                    request_context,
-                    &adapter_response,
-                )
-                .await;
-                self.store_idempotent_adapter_response(
-                    operation,
-                    idempotency_key.as_deref(),
-                    request_context,
-                    &adapter_response,
-                )
-                .await;
+                    .await;
+                if let Ok(response) = &adapter_response {
+                    self.store_cached_adapter_response(
+                        operation,
+                        &prepared_request,
+                        request_context,
+                        response,
+                    )
+                    .await;
+                }
                 adapter_response
             }
         };
-        let finalized_output = finalize_output(operation, &adapter_response)?;
-
-        operation.output_schema.validate_shape(&finalized_output)?;
-
-        Ok(finalized_output)
+        let adapter_response = match adapter_result {
+            Ok(response) => response,
+            Err(error) => {
+                if let crate::idempotency::IdempotencyAction::Execute(reservation) = &idempotency
+                    && let Some(store) = self.coordination_store.as_deref()
+                {
+                    let idempotency_span = Stage::RuntimeIdempotency.span();
+                    let cleanup_result =
+                        crate::idempotency::mark_outcome_unknown(store, operation, reservation)
+                            .instrument(idempotency_span.clone())
+                            .await;
+                    record_runtime_result(&idempotency_span, &cleanup_result);
+                }
+                return Err(error);
+            }
+        };
+        if let crate::idempotency::IdempotencyAction::Execute(reservation) = &idempotency
+            && let Some(store) = self.coordination_store.as_deref()
+        {
+            let idempotency_span = Stage::RuntimeIdempotency.span();
+            let completion_result =
+                crate::idempotency::complete(store, operation, reservation, &adapter_response)
+                    .instrument(idempotency_span.clone())
+                    .await;
+            record_runtime_result(&idempotency_span, &completion_result);
+            drop(idempotency_span);
+            completion_result?;
+        }
+        transform_response(operation, &adapter_response)
     }
 
     async fn record_metering<T>(
@@ -323,7 +411,6 @@ impl RuntimeExecutor {
         let prepared_request = adapter_prepared_request(
             operation,
             &prepared_request,
-            request_context,
             operation.execution_config.timeout_ms,
         );
         let adapter_context = adapter_request_context(request_context);
@@ -360,11 +447,11 @@ impl RuntimeExecutor {
         let cache_key = response_cache_key(operation, prepared_request, request_context)?;
         let cached = match response_cache.get(&cache_key).await {
             Ok(cached) => cached?,
-            Err(error) => {
+            Err(_) => {
                 debug!(
-                    operation_id = %operation.operation_id,
-                    cache_key,
-                    error = %error,
+                    name: "runtime.response_cache.read_failed",
+                    operation_id = operation.operation_id.as_str(),
+                    error_category = "response_cache",
                     "response cache lookup skipped"
                 );
                 return None;
@@ -373,11 +460,11 @@ impl RuntimeExecutor {
 
         match adapter_response_from_cached(cached) {
             Ok(response) => Some(response),
-            Err(error) => {
+            Err(_) => {
                 debug!(
-                    operation_id = %operation.operation_id,
-                    cache_key,
-                    error,
+                    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;
@@ -412,87 +499,19 @@ impl RuntimeExecutor {
             return;
         };
 
-        if let Err(error) = response_cache
+        if response_cache
             .put(&cache_key, cached_response, cache_ttl)
             .await
+            .is_err()
         {
             debug!(
-                operation_id = %operation.operation_id,
-                cache_key,
-                error = %error,
+                name: "runtime.response_cache.write_failed",
+                operation_id = operation.operation_id.as_str(),
+                error_category = "response_cache",
                 "response cache write skipped"
             );
         }
     }
-
-    async fn load_idempotent_adapter_response(
-        &self,
-        operation: &RuntimeOperation,
-        idempotency_key: Option<&str>,
-        request_context: Option<&RuntimeRequestContext>,
-    ) -> Option<AdapterResponse> {
-        let response_cache = self.response_cache.as_ref()?;
-        let cache_key =
-            crate::idempotency::cache_key(operation, idempotency_key?, request_context)?;
-        let cached = match response_cache.get(&cache_key).await {
-            Ok(cached) => cached?,
-            Err(error) => {
-                debug!(
-                    operation_id = %operation.operation_id,
-                    cache_key,
-                    error = %error,
-                    "idempotency cache lookup skipped"
-                );
-                return None;
-            }
-        };
-
-        adapter_response_from_cached(cached).ok()
-    }
-
-    async fn store_idempotent_adapter_response(
-        &self,
-        operation: &RuntimeOperation,
-        idempotency_key: Option<&str>,
-        request_context: Option<&RuntimeRequestContext>,
-        adapter_response: &AdapterResponse,
-    ) {
-        let Some(response_cache) = self.response_cache.as_ref() else {
-            return;
-        };
-        if !(200..=299).contains(&adapter_response.status_code) {
-            return;
-        }
-        let Some(policy) = crate::idempotency::policy(operation) else {
-            return;
-        };
-        let Some(cache_key) = crate::idempotency::cache_key(
-            operation,
-            idempotency_key.unwrap_or_default(),
-            request_context,
-        ) else {
-            return;
-        };
-        let Some(cached_response) = cached_response_from_adapter(adapter_response) else {
-            return;
-        };
-
-        if let Err(error) = response_cache
-            .put(
-                &cache_key,
-                cached_response,
-                Duration::from_millis(policy.ttl_ms),
-            )
-            .await
-        {
-            debug!(
-                operation_id = %operation.operation_id,
-                cache_key,
-                error = %error,
-                "idempotency cache write skipped"
-            );
-        }
-    }
 }
 
 fn adapter_request_context(
@@ -552,14 +571,149 @@ fn finalize_output(
         .unwrap_or_else(|| Value::Object(Map::new())))
 }
 
+fn transform_response(
+    operation: &RuntimeOperation,
+    response: &AdapterResponse,
+) -> Result<Value, RuntimeError> {
+    let span = Stage::RuntimeResponseTransform.span();
+    let result = span.in_scope(|| {
+        let finalized_output = finalize_output(operation, response)?;
+        operation.output_schema.validate_shape(&finalized_output)?;
+        Ok(finalized_output)
+    });
+    match &result {
+        Ok(_) => StageOutcome::Success.record(&span),
+        Err(_) => {
+            StageOutcome::Error.record(&span);
+            ErrorCategory::Transformation.record(&span);
+        }
+    }
+    result
+}
+
+fn record_runtime_result<T>(span: &Span, result: &Result<T, RuntimeError>) {
+    match result {
+        Ok(_) => StageOutcome::Success.record(span),
+        Err(error) => {
+            StageOutcome::Error.record(span);
+            runtime_error_category(error).record(span);
+        }
+    }
+}
+
+fn runtime_error_category(error: &RuntimeError) -> ErrorCategory {
+    match error {
+        RuntimeError::Schema(_) => ErrorCategory::Schema,
+        RuntimeError::Mapping(_) | RuntimeError::InvalidPreparedRequest { .. } => {
+            ErrorCategory::Mapping
+        }
+        RuntimeError::RestAdapter(_)
+        | RuntimeError::ProtocolAdapter(_)
+        | RuntimeError::UnsupportedProtocol { .. }
+        | RuntimeError::UnsupportedExecutionMode { .. } => ErrorCategory::Upstream,
+        RuntimeError::ConcurrencyLimitExceeded { .. } => ErrorCategory::Concurrency,
+        RuntimeError::ConfirmationRequired { .. }
+        | RuntimeError::InvalidConfirmationToken { .. }
+        | RuntimeError::ConfirmationStoreUnavailable { .. } => ErrorCategory::Approval,
+        RuntimeError::IdempotencyStoreUnavailable { .. }
+        | RuntimeError::IdempotencyInProgress { .. }
+        | RuntimeError::IdempotencyConflict { .. }
+        | RuntimeError::IdempotencyOutcomeUnknown { .. } => ErrorCategory::Idempotency,
+        RuntimeError::MissingAuthProfile { .. }
+        | RuntimeError::MissingSecret { .. }
+        | RuntimeError::MissingSecretVersion { .. }
+        | RuntimeError::InvalidAuthSecretValue { .. }
+        | RuntimeError::SecretCrypto { .. } => ErrorCategory::Configuration,
+    }
+}
+
 fn try_acquire_limit(
     limiter: Arc<Semaphore>,
     kind: &'static str,
     limit: usize,
 ) -> Result<OwnedSemaphorePermit, RuntimeError> {
-    limiter
-        .try_acquire_owned()
-        .map_err(|_| RuntimeError::ConcurrencyLimitExceeded { kind, limit })
+    limiter.try_acquire_owned().map_err(|_| {
+        metrics::counter!(
+            "crank_runtime_limit_rejections_total",
+            "stage" => "concurrency"
+        )
+        .increment(1);
+        RuntimeError::ConcurrencyLimitExceeded { kind, limit }
+    })
+}
+
+fn record_execution_metrics<T>(
+    request_context: Option<&RuntimeRequestContext>,
+    result: &Result<T, RuntimeError>,
+    started_at: Instant,
+) {
+    let source = 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());
+}
+
+fn runtime_error_kind(error: &RuntimeError) -> &'static str {
+    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",
+    }
+}
+
+struct RuntimeInFlightGuard {
+    gauge: Gauge,
+}
+
+impl RuntimeInFlightGuard {
+    fn new() -> Self {
+        let gauge = metrics::gauge!("crank_runtime_inflight");
+        gauge.increment(1.0);
+        Self { gauge }
+    }
+}
+
+impl Drop for RuntimeInFlightGuard {
+    fn drop(&mut self) {
+        self.gauge.decrement(1.0);
+    }
 }
 
 fn log_runtime_event(
@@ -567,19 +721,26 @@ fn log_runtime_event(
     operation: &RuntimeOperation,
     request_context: Option<&RuntimeRequestContext>,
 ) {
-    let request_id = request_context
-        .map(|context| context.request_id.as_str())
-        .unwrap_or_default();
-    let correlation_id = request_context
-        .map(|context| context.correlation_id.as_str())
-        .unwrap_or_default();
-
-    debug!(
-        stage,
-        operation_id = %operation.operation_id,
-        protocol = ?operation.protocol,
-        request_id,
-        correlation_id,
-        "runtime execution"
-    );
+    let protocol = match operation.protocol {
+        crank_core::Protocol::Rest => "rest",
+    };
+    if let Some(context) = request_context {
+        debug!(
+            name: "runtime.execution.stage_reached",
+            stage,
+            operation_id = operation.operation_id.as_str(),
+            protocol,
+            request_id = context.request_id.as_str(),
+            correlation_id = context.correlation_id.as_str(),
+            "runtime execution"
+        );
+    } else {
+        debug!(
+            name: "runtime.execution.stage_reached",
+            stage,
+            operation_id = operation.operation_id.as_str(),
+            protocol,
+            "runtime execution"
+        );
+    }
 }

crates/crank-runtime/src/idempotency.rs

@@ -1,9 +1,32 @@
-use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
-use crank_core::{HttpMethod, IdempotencyMode, IdempotencyPolicy, Target};
-use serde_json::Value;
-use sha2::{Digest, Sha256};
+use std::time::{Duration, Instant};
 
-use crate::{PreparedRequest, RuntimeError, RuntimeOperation, RuntimeRequestContext};
+use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
+use crank_core::{
+    CacheScope, CoordinationStateReservation, CoordinationStateStore, CoordinationStateValue,
+    HttpMethod, IdempotencyMode, IdempotencyPolicy, Target,
+};
+use serde_json::{Map, Value, json};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+use crate::{
+    AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation, RuntimeRequestContext,
+};
+
+const POLL_INTERVAL: Duration = Duration::from_millis(10);
+
+pub(crate) enum IdempotencyAction {
+    Disabled,
+    Execute(IdempotencyReservation),
+    Replay(AdapterResponse),
+}
+
+pub(crate) struct IdempotencyReservation {
+    key: String,
+    initial: CoordinationStateValue,
+    fingerprint: String,
+    result_ttl: Duration,
+}
 
 pub fn prepare_idempotency(
     operation: &RuntimeOperation,
@@ -39,6 +62,185 @@ pub fn prepare_idempotency(
     Ok(Some(key))
 }
 
+pub(crate) async fn begin(
+    store: Option<&dyn CoordinationStateStore>,
+    operation: &RuntimeOperation,
+    input: &Value,
+    idempotency_key: Option<&str>,
+    request_context: Option<&RuntimeRequestContext>,
+) -> Result<IdempotencyAction, RuntimeError> {
+    let Some(policy) = policy(operation) else {
+        return Ok(IdempotencyAction::Disabled);
+    };
+    let Some(idempotency_key) = idempotency_key else {
+        return Ok(IdempotencyAction::Disabled);
+    };
+    let Some(key) = cache_key(operation, idempotency_key, request_context) else {
+        return Err(RuntimeError::IdempotencyStoreUnavailable {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        });
+    };
+    let Some(store) = store else {
+        return Err(RuntimeError::IdempotencyStoreUnavailable {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        });
+    };
+
+    let fingerprint = request_fingerprint(input)?;
+    let result_ttl = Duration::from_millis(policy.ttl_ms);
+    let reservation_ttl = result_ttl.max(Duration::from_millis(
+        operation.execution_config.timeout_ms.saturating_add(1_000),
+    ));
+    let initial = in_progress_value(&fingerprint);
+    let reservation = store
+        .reserve_value(
+            CacheScope::Coordination,
+            &key,
+            initial.clone(),
+            reservation_ttl,
+        )
+        .await
+        .map_err(|_| RuntimeError::IdempotencyStoreUnavailable {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        })?;
+
+    match reservation {
+        CoordinationStateReservation::Reserved => {
+            Ok(IdempotencyAction::Execute(IdempotencyReservation {
+                key,
+                initial,
+                fingerprint,
+                result_ttl,
+            }))
+        }
+        CoordinationStateReservation::Existing(existing) => {
+            resolve_existing(store, operation, &key, &fingerprint, existing).await
+        }
+    }
+}
+
+pub(crate) async fn complete(
+    store: &dyn CoordinationStateStore,
+    operation: &RuntimeOperation,
+    reservation: &IdempotencyReservation,
+    response: &AdapterResponse,
+) -> Result<(), RuntimeError> {
+    let completed = CoordinationStateValue {
+        payload: json!({
+            "state": "completed",
+            "fingerprint": reservation.fingerprint,
+            "response": response,
+        }),
+    };
+    let replaced = store
+        .compare_and_set_value(
+            CacheScope::Coordination,
+            &reservation.key,
+            &reservation.initial,
+            completed,
+            reservation.result_ttl,
+        )
+        .await
+        .map_err(|_| RuntimeError::IdempotencyStoreUnavailable {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        })?;
+    if replaced {
+        Ok(())
+    } else {
+        Err(RuntimeError::IdempotencyOutcomeUnknown {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        })
+    }
+}
+
+pub(crate) async fn mark_outcome_unknown(
+    store: &dyn CoordinationStateStore,
+    operation: &RuntimeOperation,
+    reservation: &IdempotencyReservation,
+) -> Result<(), RuntimeError> {
+    let unknown = CoordinationStateValue {
+        payload: json!({
+            "state": "outcome_unknown",
+            "fingerprint": reservation.fingerprint,
+        }),
+    };
+    let replaced = store
+        .compare_and_set_value(
+            CacheScope::Coordination,
+            &reservation.key,
+            &reservation.initial,
+            unknown,
+            reservation.result_ttl,
+        )
+        .await
+        .map_err(|_| RuntimeError::IdempotencyStoreUnavailable {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        })?;
+    if replaced {
+        Ok(())
+    } else {
+        Err(RuntimeError::IdempotencyOutcomeUnknown {
+            operation_id: operation.operation_id.as_str().to_owned(),
+        })
+    }
+}
+
+async fn resolve_existing(
+    store: &dyn CoordinationStateStore,
+    operation: &RuntimeOperation,
+    key: &str,
+    fingerprint: &str,
+    mut existing: CoordinationStateValue,
+) -> Result<IdempotencyAction, RuntimeError> {
+    let operation_id = operation.operation_id.as_str().to_owned();
+    let deadline =
+        Instant::now() + Duration::from_millis(operation.execution_config.timeout_ms.max(1));
+    loop {
+        let existing_fingerprint = existing.payload.get("fingerprint").and_then(Value::as_str);
+        if existing_fingerprint != Some(fingerprint) {
+            return Err(RuntimeError::IdempotencyConflict { operation_id });
+        }
+        match existing.payload.get("state").and_then(Value::as_str) {
+            Some("completed") => {
+                let response = existing
+                    .payload
+                    .get("response")
+                    .cloned()
+                    .and_then(|value| serde_json::from_value(value).ok())
+                    .ok_or_else(|| RuntimeError::InvalidPreparedRequest {
+                        field: "idempotency_state".to_owned(),
+                        reason: "completed idempotency state has no valid response".to_owned(),
+                    })?;
+                return Ok(IdempotencyAction::Replay(response));
+            }
+            Some("outcome_unknown") => {
+                return Err(RuntimeError::IdempotencyOutcomeUnknown { operation_id });
+            }
+            Some("in_progress") if Instant::now() < deadline => {
+                tokio::time::sleep(POLL_INTERVAL).await;
+                existing = store
+                    .get_value(CacheScope::Coordination, key)
+                    .await
+                    .map_err(|_| RuntimeError::IdempotencyStoreUnavailable {
+                        operation_id: operation_id.clone(),
+                    })?
+                    .ok_or_else(|| RuntimeError::IdempotencyOutcomeUnknown {
+                        operation_id: operation_id.clone(),
+                    })?;
+            }
+            Some("in_progress") => {
+                return Err(RuntimeError::IdempotencyInProgress { operation_id });
+            }
+            _ => {
+                return Err(RuntimeError::InvalidPreparedRequest {
+                    field: "idempotency_state".to_owned(),
+                    reason: "unknown idempotency state".to_owned(),
+                });
+            }
+        }
+    }
+}
+
 pub fn policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy> {
     let is_mutating_rest =
         matches!(&operation.target, Target::Rest(target) if target.method != HttpMethod::Get);
@@ -54,7 +256,7 @@ pub fn policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy> {
     Some(policy)
 }
 
-pub fn cache_key(
+fn cache_key(
     operation: &RuntimeOperation,
     idempotency_key: &str,
     request_context: Option<&RuntimeRequestContext>,
@@ -76,6 +278,42 @@ pub fn cache_key(
     ))
 }
 
+fn request_fingerprint(input: &Value) -> Result<String, RuntimeError> {
+    let canonical = canonical_json(input);
+    let encoded =
+        serde_json::to_vec(&canonical).map_err(|error| RuntimeError::InvalidPreparedRequest {
+            field: "idempotency_fingerprint".to_owned(),
+            reason: error.to_string(),
+        })?;
+    Ok(URL_SAFE_NO_PAD.encode(Sha256::digest(encoded)))
+}
+
+fn canonical_json(value: &Value) -> Value {
+    match value {
+        Value::Object(object) => {
+            let mut entries = object.iter().collect::<Vec<_>>();
+            entries.sort_unstable_by_key(|(key, _)| *key);
+            let mut canonical = Map::new();
+            for (key, value) in entries {
+                canonical.insert(key.clone(), canonical_json(value));
+            }
+            Value::Object(canonical)
+        }
+        Value::Array(values) => Value::Array(values.iter().map(canonical_json).collect()),
+        _ => value.clone(),
+    }
+}
+
+fn in_progress_value(fingerprint: &str) -> CoordinationStateValue {
+    CoordinationStateValue {
+        payload: json!({
+            "state": "in_progress",
+            "fingerprint": fingerprint,
+            "owner": Uuid::now_v7().simple().to_string(),
+        }),
+    }
+}
+
 fn key_from_policy(
     policy: &IdempotencyPolicy,
     input: &Value,

crates/crank-runtime/src/lib.rs

@@ -32,7 +32,8 @@ pub use executor_builder::{
 pub use limits::{RuntimeLimits, RuntimeLimitsConfigError};
 pub use model::{AdapterResponse, PreparedRequest, RuntimeOperation};
 pub use rate_limit::{
-    RateLimitRejection, RequestRateLimitConfig, RequestRateLimitConfigError, RequestRateLimiter,
+    RateLimitCheckError, RateLimitRejection, RequestRateLimitConfig, RequestRateLimitConfigError,
+    RequestRateLimiter,
 };
 pub use request_context::{MeteringContext, ResponseCacheScope, RuntimeRequestContext};
 pub use secret_crypto::SecretCrypto;

crates/crank-runtime/src/limits.rs

@@ -3,16 +3,19 @@ use std::{env, num::ParseIntError};
 use thiserror::Error;
 
 const DEFAULT_MAX_CONCURRENT_UNARY: usize = 64;
+const DEFAULT_MAX_CONCURRENT_SESSIONS: usize = 16;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct RuntimeLimits {
     pub max_concurrent_unary: usize,
+    pub max_concurrent_sessions: usize,
 }
 
 impl Default for RuntimeLimits {
     fn default() -> Self {
         Self {
             max_concurrent_unary: DEFAULT_MAX_CONCURRENT_UNARY,
+            max_concurrent_sessions: DEFAULT_MAX_CONCURRENT_SESSIONS,
         }
     }
 }
@@ -24,6 +27,10 @@ impl RuntimeLimits {
                 "CRANK_RUNTIME_MAX_CONCURRENT_UNARY",
                 DEFAULT_MAX_CONCURRENT_UNARY,
             )?,
+            max_concurrent_sessions: parse_limit(
+                "CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS",
+                DEFAULT_MAX_CONCURRENT_SESSIONS,
+            )?,
         })
     }
 }
@@ -71,6 +78,7 @@ mod tests {
         let limits = RuntimeLimits::default();
 
         assert!(limits.max_concurrent_unary > 0);
+        assert!(limits.max_concurrent_sessions > 0);
     }
 
     #[test]

crates/crank-runtime/src/rate_limit.rs

@@ -4,8 +4,9 @@ use std::{
     time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
-use crank_core::{RateLimitBucketState, RateLimitStateStore};
+use crank_core::{RateLimitDecision, RateLimitStateStore};
 use thiserror::Error;
+use tracing::warn;
 
 const STALE_KEY_TTL: Duration = Duration::from_secs(300);
 const TOKEN_SCALE: u64 = 1_000_000;
@@ -47,6 +48,12 @@ pub struct RateLimitRejection {
     pub retry_after_ms: u64,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RateLimitCheckError {
+    Rejected(RateLimitRejection),
+    StoreUnavailable,
+}
+
 #[derive(Clone)]
 pub struct RequestRateLimiter {
     config: RequestRateLimitConfig,
@@ -87,14 +94,24 @@ impl RequestRateLimiter {
         }
     }
 
-    pub async fn check(&self, key: &str) -> Result<(), RateLimitRejection> {
-        match &self.backend {
-            RequestRateLimiterBackend::Local { .. } => self.check_local_at(key, Instant::now()),
+    pub async fn check(&self, key: &str) -> Result<(), RateLimitCheckError> {
+        let result = match &self.backend {
+            RequestRateLimiterBackend::Local { .. } => self
+                .check_local_at(key, Instant::now())
+                .map_err(RateLimitCheckError::Rejected),
             RequestRateLimiterBackend::Shared { store } => {
                 self.check_shared_at(store.as_ref(), key, now_unix_ms())
                     .await
             }
+        };
+        if matches!(result, Err(RateLimitCheckError::Rejected(_))) {
+            metrics::counter!(
+                "crank_runtime_limit_rejections_total",
+                "stage" => "rate_limit"
+            )
+            .increment(1);
         }
+        result
     }
 
     fn check_local_at(&self, key: &str, now: Instant) -> Result<(), RateLimitRejection> {
@@ -137,35 +154,35 @@ impl RequestRateLimiter {
         store: &dyn RateLimitStateStore,
         key: &str,
         now_unix_ms: i64,
-    ) -> Result<(), RateLimitRejection> {
-        let burst_tokens = u64::from(self.config.burst) * TOKEN_SCALE;
-        let refill_per_second = u64::from(self.config.requests_per_second) * TOKEN_SCALE;
-        let mut state =
-            store
-                .get_bucket(key)
-                .await
-                .unwrap_or(None)
-                .unwrap_or(RateLimitBucketState {
-                    tokens_micros: burst_tokens,
-                    last_refill_unix_ms: now_unix_ms,
-                });
-
-        let elapsed_ms = (now_unix_ms - state.last_refill_unix_ms).max(0) as u64;
-        let replenished =
-            state.tokens_micros + (elapsed_ms.saturating_mul(refill_per_second) / 1000);
-        state.tokens_micros = replenished.min(burst_tokens);
-        state.last_refill_unix_ms = now_unix_ms;
-
-        if state.tokens_micros >= TOKEN_SCALE {
-            state.tokens_micros -= TOKEN_SCALE;
-            let _ = store.put_bucket(key, state, STALE_KEY_TTL).await;
-            return Ok(());
+    ) -> Result<(), RateLimitCheckError> {
+        let decision = match store
+            .consume_token(
+                key,
+                u64::from(self.config.burst) * TOKEN_SCALE,
+                u64::from(self.config.requests_per_second) * TOKEN_SCALE,
+                now_unix_ms,
+                STALE_KEY_TTL,
+            )
+            .await
+        {
+            Ok(decision) => decision,
+            Err(_) => {
+                warn!(
+                    name: "runtime.rate_limit.failed_closed",
+                    error_category = "coordination_store",
+                    "shared rate limiter failed closed"
+                );
+                return Err(RateLimitCheckError::StoreUnavailable);
+            }
+        };
+        match decision {
+            RateLimitDecision::Allowed => Ok(()),
+            RateLimitDecision::Rejected { retry_after_ms } => {
+                Err(RateLimitCheckError::Rejected(RateLimitRejection {
+                    retry_after_ms,
+                }))
+            }
         }
-
-        let missing_tokens = TOKEN_SCALE.saturating_sub(state.tokens_micros);
-        let retry_after_ms = missing_tokens.div_ceil(refill_per_second).max(1);
-        let _ = store.put_bucket(key, state, STALE_KEY_TTL).await;
-        Err(RateLimitRejection { retry_after_ms })
     }
 }
 
@@ -185,6 +202,11 @@ mod tests {
         time::{Duration, Instant},
     };
 
+    use async_trait::async_trait;
+    use crank_core::{
+        CacheStoreError, RateLimitBucketState, RateLimitDecision, RateLimitStateStore,
+    };
+
     use crate::InMemoryRateLimitStateStore;
 
     use super::{RequestRateLimitConfig, RequestRateLimitConfigError, RequestRateLimiter};
@@ -234,7 +256,12 @@ mod tests {
         assert!(limiter.check_shared_at_store("key", 0).await.is_ok());
 
         let rejection = limiter.check_shared_at_store("key", 0).await.unwrap_err();
-        assert_eq!(rejection.retry_after_ms, 1);
+        assert_eq!(
+            rejection,
+            super::RateLimitCheckError::Rejected(super::RateLimitRejection {
+                retry_after_ms: 500,
+            })
+        );
 
         assert!(limiter.check_shared_at_store("key", 500).await.is_ok());
     }
@@ -254,12 +281,120 @@ mod tests {
         assert!(second.check_shared_at_store("shared", 1000).await.is_ok());
     }
 
+    #[tokio::test]
+    async fn shared_limiter_does_not_refill_when_clock_moves_backwards() {
+        let limiter = RequestRateLimiter::new_shared(
+            RequestRateLimitConfig::new(1, 1).unwrap(),
+            Arc::new(InMemoryRateLimitStateStore::default()),
+        );
+
+        assert!(limiter.check_shared_at_store("clock", 1_000).await.is_ok());
+        assert_eq!(
+            limiter
+                .check_shared_at_store("clock", 500)
+                .await
+                .unwrap_err(),
+            super::RateLimitCheckError::Rejected(super::RateLimitRejection {
+                retry_after_ms: 1_000,
+            })
+        );
+        assert_eq!(
+            limiter
+                .check_shared_at_store("clock", 1_500)
+                .await
+                .unwrap_err(),
+            super::RateLimitCheckError::Rejected(super::RateLimitRejection {
+                retry_after_ms: 500,
+            })
+        );
+        assert!(limiter.check_shared_at_store("clock", 2_000).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn shared_limiter_consumes_burst_atomically_under_concurrency() {
+        let limiter = RequestRateLimiter::new_shared(
+            RequestRateLimitConfig::new(1, 8).unwrap(),
+            Arc::new(InMemoryRateLimitStateStore::default()),
+        );
+        let attempts = (0..64).map(|_| limiter.check_shared_at_store("concurrent", 0));
+        let results = futures_util::future::join_all(attempts).await;
+
+        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 8);
+        assert!(
+            results
+                .iter()
+                .filter_map(|result| result.as_ref().err())
+                .all(|rejection| {
+                    *rejection
+                        == super::RateLimitCheckError::Rejected(super::RateLimitRejection {
+                            retry_after_ms: 1_000,
+                        })
+                })
+        );
+    }
+
+    #[tokio::test]
+    async fn shared_limiter_fails_closed_when_store_is_unavailable() {
+        let limiter = RequestRateLimiter::new_shared(
+            RequestRateLimitConfig::new(10, 10).unwrap(),
+            Arc::new(UnavailableRateLimitStore),
+        );
+
+        let error = limiter
+            .check_shared_at_store("unavailable", 0)
+            .await
+            .unwrap_err();
+        assert_eq!(error, super::RateLimitCheckError::StoreUnavailable);
+    }
+
+    struct UnavailableRateLimitStore;
+
+    #[async_trait]
+    impl RateLimitStateStore for UnavailableRateLimitStore {
+        async fn get_bucket(
+            &self,
+            _key: &str,
+        ) -> Result<Option<RateLimitBucketState>, CacheStoreError> {
+            Err(unavailable())
+        }
+
+        async fn put_bucket(
+            &self,
+            _key: &str,
+            _value: RateLimitBucketState,
+            _ttl: Duration,
+        ) -> Result<(), CacheStoreError> {
+            Err(unavailable())
+        }
+
+        async fn delete_bucket(&self, _key: &str) -> Result<(), CacheStoreError> {
+            Err(unavailable())
+        }
+
+        async fn consume_token(
+            &self,
+            _key: &str,
+            _burst_tokens_micros: u64,
+            _refill_per_second_micros: u64,
+            _now_unix_ms: i64,
+            _ttl: Duration,
+        ) -> Result<RateLimitDecision, CacheStoreError> {
+            Err(unavailable())
+        }
+    }
+
+    fn unavailable() -> CacheStoreError {
+        CacheStoreError::Unavailable {
+            message: "test store is unavailable".to_owned(),
+        }
+    }
+
     impl RequestRateLimiter {
         async fn check_shared_at_store(
             &self,
             key: &str,
             now_unix_ms: i64,
-        ) -> Result<(), super::RateLimitRejection> {
+        ) -> Result<(), super::RateLimitCheckError> {
             let super::RequestRateLimiterBackend::Shared { store } = &self.backend else {
                 panic!("check_shared_at_store called for non-shared limiter");
             };

crates/crank-runtime/src/request_preparation.rs

@@ -3,7 +3,7 @@ use std::collections::BTreeMap;
 use crank_core::Target;
 use serde_json::Value;
 
-use crate::{PreparedRequest, RuntimeError, RuntimeOperation, RuntimeRequestContext};
+use crate::{PreparedRequest, RuntimeError, RuntimeOperation};
 
 impl PreparedRequest {
     pub fn from_mapping_output(mapped: &Value) -> Result<Self, RuntimeError> {
@@ -28,7 +28,6 @@ impl PreparedRequest {
 pub(crate) fn adapter_prepared_request(
     operation: &RuntimeOperation,
     prepared_request: &PreparedRequest,
-    request_context: Option<&RuntimeRequestContext>,
     timeout_ms: u64,
 ) -> PreparedRequest {
     let static_headers = match &operation.target {
@@ -40,7 +39,6 @@ pub(crate) fn adapter_prepared_request(
         static_headers,
         &operation.execution_config.headers,
         &prepared_request.headers,
-        &runtime_context_headers(request_context),
     );
     prepared_request.timeout_ms = timeout_ms;
     prepared_request
@@ -50,23 +48,13 @@ fn merge_headers(
     static_headers: &BTreeMap<String, String>,
     execution_headers: &BTreeMap<String, String>,
     request_headers: &BTreeMap<String, String>,
-    context_headers: &BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
     let mut headers = static_headers.clone();
     headers.extend(execution_headers.clone());
     headers.extend(request_headers.clone());
-    headers.extend(context_headers.clone());
     headers
 }
 
-fn runtime_context_headers(
-    request_context: Option<&RuntimeRequestContext>,
-) -> BTreeMap<String, String> {
-    request_context
-        .map(RuntimeRequestContext::outbound_headers)
-        .unwrap_or_default()
-}
-
 fn read_string_map(
     value: Option<&Value>,
     field_name: &str,

crates/crank-runtime/tests/integration.rs

@@ -2,4 +2,6 @@ mod integration {
     mod confirmation;
     mod idempotency;
     mod no_input_get;
+    mod stages;
+    mod valkey;
 }

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

@@ -16,6 +16,7 @@ use crank_runtime::{
     InMemoryCoordinationStateStore, RuntimeError, RuntimeExecutorBuilder, RuntimeRequestContext,
 };
 use crank_schema::{Schema, SchemaKind};
+use futures_util::future::join_all;
 use serde_json::json;
 use time::OffsetDateTime;
 
@@ -89,6 +90,62 @@ async fn destructive_operation_requires_single_use_confirmation() {
     assert_eq!(call_count.load(Ordering::SeqCst), 2);
 }
 
+#[tokio::test]
+async fn confirmation_token_allows_only_one_concurrent_execution() {
+    let call_count = Arc::new(AtomicUsize::new(0));
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(CountingAdapter {
+            call_count: Arc::clone(&call_count),
+        }))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let operation: crank_runtime::RuntimeOperation = destructive_delete_operation().into();
+    let context = RuntimeRequestContext::from_request_id("req_confirm_concurrent")
+        .with_response_cache_scope("workspace_1", "agent_1");
+    let first = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "order_id": "ord_123" }),
+            Some(&context),
+        )
+        .await
+        .unwrap_err();
+    let RuntimeError::ConfirmationRequired {
+        confirmation_token, ..
+    } = first
+    else {
+        panic!("expected confirmation token")
+    };
+
+    let attempts = (0..16).map(|_| {
+        let executor = executor.clone();
+        let operation = operation.clone();
+        let context = context
+            .clone()
+            .with_confirmation_token(confirmation_token.clone());
+        async move {
+            executor
+                .execute_with_context(
+                    &operation,
+                    &json!({ "order_id": "ord_123" }),
+                    Some(&context),
+                )
+                .await
+        }
+    });
+    let results = join_all(attempts).await;
+    let successful = results.iter().filter(|result| result.is_ok()).count();
+
+    assert_eq!(successful, 1);
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+    assert!(
+        results
+            .iter()
+            .filter_map(|result| result.as_ref().err())
+            .all(|error| matches!(error, RuntimeError::InvalidConfirmationToken { .. }))
+    );
+}
+
 struct CountingAdapter {
     call_count: Arc<AtomicUsize>,
 }

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

@@ -12,7 +12,10 @@ use crank_core::{
     ToolDescription,
 };
 use crank_mapping::{MappingRule, MappingSet};
-use crank_runtime::{InMemoryResponseCacheStore, RuntimeExecutorBuilder};
+use crank_runtime::{
+    InMemoryCoordinationStateStore, InMemoryResponseCacheStore, RuntimeError,
+    RuntimeExecutorBuilder,
+};
 use crank_schema::{Schema, SchemaKind};
 use serde_json::json;
 use time::OffsetDateTime;
@@ -23,8 +26,10 @@ async fn replays_mutation_result_for_same_idempotency_key() {
     let executor = RuntimeExecutorBuilder::new()
         .register_adapter(Arc::new(CountingAdapter {
             call_count: Arc::clone(&call_count),
+            release: None,
         }))
         .with_response_cache(Arc::new(InMemoryResponseCacheStore::default()))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
         .build();
     let operation = idempotent_post_operation().into();
     let context = crank_runtime::RuntimeRequestContext::from_request_id("req_1")
@@ -51,8 +56,10 @@ async fn required_idempotency_rejects_missing_key_before_adapter_call() {
     let executor = RuntimeExecutorBuilder::new()
         .register_adapter(Arc::new(CountingAdapter {
             call_count: Arc::clone(&call_count),
+            release: None,
         }))
         .with_response_cache(Arc::new(InMemoryResponseCacheStore::default()))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
         .build();
     let mut operation: crank_runtime::RuntimeOperation = idempotent_post_operation().into();
     let policy = operation.execution_config.idempotency.as_mut().unwrap();
@@ -71,8 +78,189 @@ async fn required_idempotency_rejects_missing_key_before_adapter_call() {
     assert_eq!(call_count.load(Ordering::SeqCst), 0);
 }
 
+#[tokio::test]
+async fn required_idempotency_fails_closed_without_coordination_store() {
+    let call_count = Arc::new(AtomicUsize::new(0));
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(CountingAdapter {
+            call_count: Arc::clone(&call_count),
+            release: None,
+        }))
+        .build();
+    let operation = idempotent_post_operation().into();
+    let context = crank_runtime::RuntimeRequestContext::from_request_id("req_no_store")
+        .with_response_cache_scope("workspace_1", "agent_1");
+
+    let error = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "request_id": "must-not-run" }),
+            Some(&context),
+        )
+        .await
+        .expect_err("required idempotency must not execute without an atomic store");
+
+    assert!(matches!(
+        error,
+        RuntimeError::IdempotencyStoreUnavailable { .. }
+    ));
+    assert_eq!(call_count.load(Ordering::SeqCst), 0);
+}
+
+#[tokio::test]
+async fn concurrent_calls_with_same_key_execute_adapter_once() {
+    let call_count = Arc::new(AtomicUsize::new(0));
+    let release = Arc::new(tokio::sync::Notify::new());
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(CountingAdapter {
+            call_count: Arc::clone(&call_count),
+            release: Some(Arc::clone(&release)),
+        }))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let operation: crank_runtime::RuntimeOperation = idempotent_post_operation().into();
+    let context = crank_runtime::RuntimeRequestContext::from_request_id("req_concurrent")
+        .with_response_cache_scope("workspace_1", "agent_1");
+    let input = json!({ "request_id": "order-concurrent" });
+
+    let first_executor = executor.clone();
+    let first_operation = operation.clone();
+    let first_context = context.clone();
+    let first_input = input.clone();
+    let first = tokio::spawn(async move {
+        first_executor
+            .execute_with_context(&first_operation, &first_input, Some(&first_context))
+            .await
+    });
+    wait_for_call_count(&call_count, 1).await;
+
+    let second_executor = executor.clone();
+    let second_operation = operation.clone();
+    let second_context = context.clone();
+    let second_input = input.clone();
+    let second = tokio::spawn(async move {
+        second_executor
+            .execute_with_context(&second_operation, &second_input, Some(&second_context))
+            .await
+    });
+    for _ in 0..100 {
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+
+    release.notify_waiters();
+    let first_result = first.await.unwrap().unwrap();
+    let second_result = second.await.unwrap().unwrap();
+    assert_eq!(first_result, second_result);
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+}
+
+#[tokio::test]
+async fn same_key_with_different_input_is_rejected() {
+    let call_count = Arc::new(AtomicUsize::new(0));
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(CountingAdapter {
+            call_count: Arc::clone(&call_count),
+            release: None,
+        }))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let operation = idempotent_post_operation().into();
+    let context = crank_runtime::RuntimeRequestContext::from_request_id("req_conflict")
+        .with_response_cache_scope("workspace_1", "agent_1");
+
+    executor
+        .execute_with_context(
+            &operation,
+            &json!({ "request_id": "stable-key", "amount": 10 }),
+            Some(&context),
+        )
+        .await
+        .unwrap();
+    let error = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "request_id": "stable-key", "amount": 20 }),
+            Some(&context),
+        )
+        .await
+        .expect_err("same key must not accept a different request fingerprint");
+
+    assert!(matches!(error, RuntimeError::IdempotencyConflict { .. }));
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+}
+
+#[tokio::test]
+async fn uncertain_adapter_failure_blocks_automatic_retry() {
+    let call_count = Arc::new(AtomicUsize::new(0));
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(FailingAdapter {
+            call_count: Arc::clone(&call_count),
+        }))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let operation = idempotent_post_operation().into();
+    let context = crank_runtime::RuntimeRequestContext::from_request_id("req_unknown")
+        .with_response_cache_scope("workspace_1", "agent_1");
+    let input = json!({ "request_id": "uncertain-outcome" });
+
+    let first = executor
+        .execute_with_context(&operation, &input, Some(&context))
+        .await
+        .expect_err("adapter failure must be returned");
+    assert!(matches!(first, RuntimeError::ProtocolAdapter(_)));
+
+    let retry = executor
+        .execute_with_context(&operation, &input, Some(&context))
+        .await
+        .expect_err("an uncertain external outcome must not be retried automatically");
+    assert!(matches!(
+        retry,
+        RuntimeError::IdempotencyOutcomeUnknown { .. }
+    ));
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+}
+
+async fn wait_for_call_count(call_count: &AtomicUsize, expected: usize) {
+    for _ in 0..1_000 {
+        if call_count.load(Ordering::SeqCst) >= expected {
+            return;
+        }
+        tokio::task::yield_now().await;
+    }
+    panic!("adapter did not receive {expected} call(s)");
+}
+
 struct CountingAdapter {
     call_count: Arc<AtomicUsize>,
+    release: Option<Arc<tokio::sync::Notify>>,
+}
+
+struct FailingAdapter {
+    call_count: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl ProtocolAdapter for FailingAdapter {
+    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: &RuntimeRequestContext,
+    ) -> Result<AdapterResponse, ProtocolAdapterError> {
+        self.call_count.fetch_add(1, Ordering::SeqCst);
+        Err(ProtocolAdapterError::Message(
+            "upstream outcome is unknown".to_owned(),
+        ))
+    }
 }
 
 #[async_trait]
@@ -91,11 +279,11 @@ impl ProtocolAdapter for CountingAdapter {
         prepared: &crank_core::PreparedRequest,
         _context: &RuntimeRequestContext,
     ) -> Result<AdapterResponse, ProtocolAdapterError> {
-        assert_eq!(
-            prepared.headers.get("Idempotency-Key").map(String::as_str),
-            Some("order-123")
-        );
+        assert!(prepared.headers.contains_key("Idempotency-Key"));
         let call_number = self.call_count.fetch_add(1, Ordering::SeqCst) + 1;
+        if let Some(release) = &self.release {
+            release.notified().await;
+        }
         Ok(AdapterResponse {
             status_code: 201,
             headers: BTreeMap::new(),

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

@@ -0,0 +1,426 @@
+use std::{
+    collections::BTreeMap,
+    sync::{Arc, Mutex},
+};
+
+use async_trait::async_trait;
+use crank_core::{
+    AdapterResponse, ConfirmationPolicy, ExecutionConfig, ExecutionMode, HttpMethod,
+    IdempotencyMode, IdempotencyPolicy, Operation, OperationId, OperationSafetyClass,
+    OperationSafetyPolicy, OperationSecurityLevel, OperationStatus, Protocol, ProtocolAdapter,
+    ProtocolAdapterError, RestTarget, Target, ToolDescription,
+};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_runtime::{
+    InMemoryCoordinationStateStore, RuntimeError, RuntimeExecutorBuilder, RuntimeRequestContext,
+};
+use crank_schema::{Schema, SchemaKind};
+use crank_trace::{Stage, StageOutcome};
+use serde_json::json;
+use time::OffsetDateTime;
+use tracing::{Id, Instrument, Subscriber, field::Visit, instrument::WithSubscriber};
+use tracing_subscriber::{Layer, layer::SubscriberExt, registry::LookupSpan};
+
+#[tokio::test]
+async fn successful_execution_has_real_stages_and_omits_inapplicable_ones() {
+    let capture = TraceCapture::default();
+    let subscriber = tracing_subscriber::registry().with(capture.clone());
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(SuccessAdapter))
+        .build();
+    let operation = operation().into();
+    let context = RuntimeRequestContext::from_request_id("req_stage_test");
+    let result = async {
+        let root = tracing::info_span!(target: "crank::trace", "mcp.request");
+        executor
+            .execute_with_context(
+                &operation,
+                &json!({"name": "canary-secret"}),
+                Some(&context),
+            )
+            .instrument(root)
+            .await
+    }
+    .with_subscriber(subscriber)
+    .await;
+
+    assert_eq!(result.unwrap(), json!({"accepted": true}));
+    let spans = capture.snapshot();
+    assert_stage(&spans, "runtime.execute", "success");
+    assert_stage(&spans, "runtime.arguments.map", "success");
+    assert_stage(&spans, "upstream.http", "success");
+    assert_stage(&spans, "runtime.response.transform", "success");
+    assert!(!spans.iter().any(|span| span.name == "approval.check"));
+    assert!(!spans.iter().any(|span| span.name == "runtime.idempotency"));
+    assert!(
+        spans
+            .iter()
+            .flat_map(|span| span.fields.values())
+            .all(|value| !value.contains("canary-secret"))
+    );
+
+    let runtime = spans
+        .iter()
+        .find(|span| span.name == "runtime.execute")
+        .expect("runtime span");
+    assert_eq!(runtime.parent_name, Some("mcp.request"));
+    for child in [
+        "runtime.arguments.map",
+        "upstream.http",
+        "runtime.response.transform",
+    ] {
+        assert_eq!(
+            spans
+                .iter()
+                .find(|span| span.name == child)
+                .and_then(|span| span.parent_name),
+            Some("runtime.execute"),
+            "{child} must be a runtime child"
+        );
+    }
+}
+
+#[tokio::test]
+async fn failed_mapping_records_closed_category_and_stops_later_stages() {
+    let capture = TraceCapture::default();
+    let subscriber = tracing_subscriber::registry().with(capture.clone());
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(SuccessAdapter))
+        .build();
+    let operation = operation().into();
+
+    let result = async { executor.execute(&operation, &json!({})).await }
+        .with_subscriber(subscriber)
+        .await;
+
+    assert!(result.is_err());
+    let spans = capture.snapshot();
+    let runtime = spans
+        .iter()
+        .find(|span| span.name == "runtime.execute")
+        .expect("runtime span");
+    assert_eq!(runtime.fields["outcome"], "error");
+    assert_eq!(runtime.fields["error.category"], "schema");
+    assert_stage(&spans, "runtime.arguments.map", "error");
+    assert!(!spans.iter().any(|span| span.name == "upstream.http"));
+    assert!(
+        !spans
+            .iter()
+            .any(|span| span.name == "runtime.response.transform")
+    );
+}
+
+#[tokio::test]
+async fn approval_stage_is_present_only_when_confirmation_is_required() {
+    let capture = TraceCapture::default();
+    let subscriber = tracing_subscriber::registry().with(capture.clone());
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(SuccessAdapter))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let mut source = operation();
+    source.execution_config.safety = Some(OperationSafetyPolicy {
+        class: OperationSafetyClass::Destructive,
+        confirmation: Some(ConfirmationPolicy { ttl_ms: 60_000 }),
+    });
+    let operation = source.into();
+    let context = RuntimeRequestContext::from_request_id("req_approval_stage")
+        .with_response_cache_scope("workspace", "agent");
+
+    let result = async {
+        executor
+            .execute_with_context(
+                &operation,
+                &json!({"name": "requires-confirmation"}),
+                Some(&context),
+            )
+            .await
+    }
+    .with_subscriber(subscriber)
+    .await;
+
+    assert!(matches!(
+        result,
+        Err(RuntimeError::ConfirmationRequired { .. })
+    ));
+    let spans = capture.snapshot();
+    assert_stage(&spans, "approval.check", "required");
+    assert!(!spans.iter().any(|span| span.name == "upstream.http"));
+    assert!(!spans.iter().any(|span| span.name == "runtime.idempotency"));
+}
+
+#[tokio::test]
+async fn idempotency_stage_distinguishes_execution_from_replay() {
+    let capture = TraceCapture::default();
+    let subscriber = tracing_subscriber::registry().with(capture.clone());
+    let executor = RuntimeExecutorBuilder::new()
+        .register_adapter(Arc::new(SuccessAdapter))
+        .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default()))
+        .build();
+    let mut source = operation();
+    source.execution_config.idempotency = Some(IdempotencyPolicy {
+        mode: IdempotencyMode::Required,
+        ttl_ms: 60_000,
+        input_field: Some("name".to_owned()),
+        header_name: Some("Idempotency-Key".to_owned()),
+    });
+    let operation = source.into();
+    let context = RuntimeRequestContext::from_request_id("req_idempotency_stage")
+        .with_response_cache_scope("workspace", "agent");
+
+    let (first, replay) = async {
+        let first = executor
+            .execute_with_context(&operation, &json!({"name": "stable-key"}), Some(&context))
+            .await;
+        let replay = executor
+            .execute_with_context(&operation, &json!({"name": "stable-key"}), Some(&context))
+            .await;
+        (first, replay)
+    }
+    .with_subscriber(subscriber)
+    .await;
+
+    assert!(first.is_ok());
+    assert!(replay.is_ok());
+    let spans = capture.snapshot();
+    let idempotency_outcomes = spans
+        .iter()
+        .filter(|span| span.name == "runtime.idempotency")
+        .map(|span| span.fields["outcome"].as_str())
+        .collect::<Vec<_>>();
+    assert!(idempotency_outcomes.contains(&"execute"));
+    assert!(idempotency_outcomes.contains(&"replay"));
+    assert_eq!(
+        spans
+            .iter()
+            .filter(|span| span.name == "upstream.http")
+            .count(),
+        1,
+        "replay must not pretend to call upstream"
+    );
+}
+
+fn assert_stage(spans: &[CapturedSpan], name: &str, outcome: &str) {
+    let span = spans
+        .iter()
+        .find(|span| span.name == name)
+        .unwrap_or_else(|| panic!("missing stage {name}"));
+    assert_eq!(span.fields["outcome"], outcome);
+}
+
+struct SuccessAdapter;
+
+#[async_trait]
+impl ProtocolAdapter for SuccessAdapter {
+    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: &crank_core::RuntimeRequestContext,
+    ) -> Result<AdapterResponse, ProtocolAdapterError> {
+        let span = Stage::UpstreamHttp.span();
+        let response = Ok(AdapterResponse {
+            status_code: 200,
+            headers: BTreeMap::new(),
+            body: json!({"accepted": true}),
+            data: json!({"accepted": true}),
+        });
+        StageOutcome::Success.record(&span);
+        response
+    }
+}
+
+fn operation() -> Operation<Schema, MappingSet> {
+    Operation {
+        id: OperationId::new("op_stage_test"),
+        name: "stage_test".to_owned(),
+        display_name: "Stage test".to_owned(),
+        category: "test".to_owned(),
+        protocol: Protocol::Rest,
+        security_level: OperationSecurityLevel::Standard,
+        status: OperationStatus::Published,
+        version: 1,
+        target: Target::Rest(RestTarget {
+            base_url: "https://example.invalid".to_owned(),
+            method: HttpMethod::Post,
+            path_template: "/test".to_owned(),
+            static_headers: BTreeMap::new(),
+        }),
+        input_schema: object_schema(BTreeMap::from([("name".to_owned(), string_schema())])),
+        output_schema: object_schema(BTreeMap::from([("accepted".to_owned(), bool_schema())])),
+        input_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.mcp.name".to_owned(),
+                target: "$.request.body.name".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        output_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.response.body.accepted".to_owned(),
+                target: "$.output.accepted".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: None,
+            idempotency: None,
+            safety: None,
+            approval_policy: None,
+            auth_profile_ref: None,
+            headers: BTreeMap::new(),
+        },
+        tool_description: ToolDescription {
+            title: "Stage test".to_owned(),
+            description: "Tests trace stages.".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(fields: BTreeMap<String, Schema>) -> Schema {
+    Schema {
+        kind: SchemaKind::Object,
+        description: None,
+        required: true,
+        nullable: false,
+        default_value: None,
+        fields,
+        items: None,
+        enum_values: Vec::new(),
+        variants: Vec::new(),
+    }
+}
+
+fn string_schema() -> Schema {
+    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(),
+    }
+}
+
+fn bool_schema() -> Schema {
+    Schema {
+        kind: SchemaKind::Boolean,
+        ..string_schema()
+    }
+}
+
+#[derive(Clone, Default)]
+struct TraceCapture {
+    spans: Arc<Mutex<Vec<CapturedSpan>>>,
+}
+
+impl TraceCapture {
+    fn snapshot(&self) -> Vec<CapturedSpan> {
+        self.spans.lock().expect("span lock").clone()
+    }
+}
+
+#[derive(Clone, Debug)]
+struct CapturedSpan {
+    name: &'static str,
+    parent_name: Option<&'static str>,
+    fields: BTreeMap<String, String>,
+}
+
+impl<S> Layer<S> for TraceCapture
+where
+    S: Subscriber + for<'lookup> LookupSpan<'lookup>,
+{
+    fn on_new_span(
+        &self,
+        attributes: &tracing::span::Attributes<'_>,
+        id: &Id,
+        context: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        let parent = attributes
+            .parent()
+            .and_then(|parent| context.span(parent))
+            .or_else(|| {
+                attributes
+                    .is_contextual()
+                    .then(|| context.lookup_current())
+                    .flatten()
+            });
+        let mut visitor = FieldVisitor::default();
+        attributes.record(&mut visitor);
+        let mut spans = self.spans.lock().expect("span lock");
+        let index = spans.len();
+        spans.push(CapturedSpan {
+            name: attributes.metadata().name(),
+            parent_name: parent.map(|span| span.metadata().name()),
+            fields: visitor.fields,
+        });
+        context
+            .span(id)
+            .expect("span exists")
+            .extensions_mut()
+            .insert(index);
+    }
+
+    fn on_record(
+        &self,
+        id: &Id,
+        values: &tracing::span::Record<'_>,
+        context: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        let mut visitor = FieldVisitor::default();
+        values.record(&mut visitor);
+        let span = context.span(id).expect("span exists");
+        let index = *span.extensions().get::<usize>().expect("capture index");
+        self.spans.lock().expect("span lock")[index]
+            .fields
+            .extend(visitor.fields);
+    }
+}
+
+#[derive(Default)]
+struct FieldVisitor {
+    fields: BTreeMap<String, String>,
+}
+
+impl Visit for FieldVisitor {
+    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+        self.fields
+            .insert(field.name().to_owned(), value.to_owned());
+    }
+
+    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+        self.fields
+            .insert(field.name().to_owned(), format!("{value:?}"));
+    }
+}

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

@@ -0,0 +1,167 @@
+use std::{sync::Arc, time::Duration};
+
+use crank_core::{
+    CacheBackend, CacheScope, CoordinationStateReservation, CoordinationStateStore,
+    CoordinationStateValue, RateLimitDecision, RateLimitStateStore,
+};
+use crank_runtime::RedisCacheStore;
+use futures_util::future::join_all;
+use serde_json::json;
+use testcontainers::{
+    GenericImage,
+    core::{IntoContainerPort, WaitFor},
+    runners::AsyncRunner,
+};
+
+#[tokio::test]
+async fn valkey_coordination_and_rate_limit_operations_are_atomic() {
+    let container = GenericImage::new("valkey/valkey", "8-alpine")
+        .with_exposed_port(6379.tcp())
+        .with_wait_for(WaitFor::message_on_stdout("Ready to accept connections"))
+        .start()
+        .await
+        .expect("Valkey test container must start");
+    let port = container
+        .get_host_port_ipv4(6379.tcp())
+        .await
+        .expect("Valkey port must be mapped");
+    let store = Arc::new(
+        RedisCacheStore::connect(CacheBackend::Valkey, &format!("redis://127.0.0.1:{port}/0"))
+            .await
+            .expect("runtime store must connect to Valkey"),
+    );
+
+    verify_atomic_coordination(store.as_ref()).await;
+    verify_atomic_rate_limit(store).await;
+}
+
+async fn verify_atomic_coordination(store: &RedisCacheStore) {
+    let pending = CoordinationStateValue {
+        payload: json!({ "state": "pending" }),
+    };
+    let attempts = (0..32).map(|_| {
+        store.reserve_value(
+            CacheScope::Coordination,
+            "valkey-reservation",
+            pending.clone(),
+            Duration::from_secs(30),
+        )
+    });
+    let results = join_all(attempts).await;
+    assert_eq!(
+        results
+            .iter()
+            .filter(|result| matches!(result, Ok(CoordinationStateReservation::Reserved)))
+            .count(),
+        1
+    );
+
+    let completed = CoordinationStateValue {
+        payload: json!({ "state": "completed" }),
+    };
+    assert!(
+        store
+            .compare_and_set_value(
+                CacheScope::Coordination,
+                "valkey-reservation",
+                &pending,
+                completed.clone(),
+                Duration::from_secs(30),
+            )
+            .await
+            .unwrap()
+    );
+    assert_eq!(
+        store
+            .take_value(CacheScope::Coordination, "valkey-reservation")
+            .await
+            .unwrap(),
+        Some(completed)
+    );
+    assert_eq!(
+        store
+            .take_value(CacheScope::Coordination, "valkey-reservation")
+            .await
+            .unwrap(),
+        None
+    );
+}
+
+async fn verify_atomic_rate_limit(store: Arc<RedisCacheStore>) {
+    let attempts = (0..64).map(|_| {
+        let store = Arc::clone(&store);
+        async move {
+            store
+                .consume_token(
+                    "valkey-burst",
+                    8_000_000,
+                    1_000_000,
+                    0,
+                    Duration::from_secs(30),
+                )
+                .await
+        }
+    });
+    let results = join_all(attempts).await;
+    assert_eq!(
+        results
+            .iter()
+            .filter(|result| matches!(result, Ok(RateLimitDecision::Allowed)))
+            .count(),
+        8
+    );
+    assert!(
+        results
+            .iter()
+            .filter_map(|result| result.as_ref().ok())
+            .all(|decision| matches!(
+                decision,
+                RateLimitDecision::Allowed
+                    | RateLimitDecision::Rejected {
+                        retry_after_ms: 1_000
+                    }
+            ))
+    );
+
+    assert_eq!(
+        store
+            .consume_token(
+                "valkey-retry-after",
+                2_000_000,
+                2_000_000,
+                0,
+                Duration::from_secs(30),
+            )
+            .await
+            .unwrap(),
+        RateLimitDecision::Allowed
+    );
+    assert_eq!(
+        store
+            .consume_token(
+                "valkey-retry-after",
+                2_000_000,
+                2_000_000,
+                0,
+                Duration::from_secs(30),
+            )
+            .await
+            .unwrap(),
+        RateLimitDecision::Allowed
+    );
+    assert_eq!(
+        store
+            .consume_token(
+                "valkey-retry-after",
+                2_000_000,
+                2_000_000,
+                0,
+                Duration::from_secs(30),
+            )
+            .await
+            .unwrap(),
+        RateLimitDecision::Rejected {
+            retry_after_ms: 500
+        }
+    );
+}

crates/crank-runtime/tests/unit/cache.rs

@@ -1,9 +1,14 @@
-use std::time::Duration;
+use std::{
+    ffi::OsString,
+    sync::{Mutex, MutexGuard},
+    time::Duration,
+};
 
 use crank_core::{
     CacheBackend, CacheScope, CacheStoreError, CachedHeader, CachedResponse,
-    CoordinationStateStore, CoordinationStateValue, RateLimitBucketState, RateLimitStateStore,
-    ReplayGuardStatus, ReplayGuardStore, ResponseCacheStore,
+    CoordinationStateReservation, CoordinationStateStore, CoordinationStateValue,
+    RateLimitBucketState, RateLimitStateStore, ReplayGuardStatus, ReplayGuardStore,
+    ResponseCacheStore,
 };
 use crank_runtime::{
     InMemoryCoordinationStateStore, InMemoryRateLimitStateStore, InMemoryReplayGuardStore,
@@ -12,6 +17,13 @@ use crank_runtime::{
 };
 use serde_json::json;
 
+const CACHE_ENV_NAMES: [&str; 3] = [
+    "CRANK_CACHE_BACKEND",
+    "CRANK_CACHE_URL",
+    "CRANK_CACHE_DEFAULT_TTL_MS",
+];
+static CACHE_ENV_LOCK: Mutex<()> = Mutex::new(());
+
 #[test]
 fn defaults_to_in_memory_cache_without_url() {
     let config = RuntimeCacheConfig::default();
@@ -21,8 +33,24 @@ fn defaults_to_in_memory_cache_without_url() {
     assert_eq!(config.default_ttl_ms, None);
 }
 
+#[test]
+fn treats_blank_optional_cache_values_as_unset() {
+    let _env = IsolatedCacheEnv::new();
+    unsafe {
+        std::env::set_var("CRANK_CACHE_URL", "  ");
+        std::env::set_var("CRANK_CACHE_DEFAULT_TTL_MS", "  ");
+    }
+
+    let config = RuntimeCacheConfig::from_env().unwrap();
+
+    assert_eq!(config.backend, CacheBackend::Memory);
+    assert_eq!(config.url, None);
+    assert_eq!(config.default_ttl_ms, None);
+}
+
 #[test]
 fn loads_valkey_config_from_env() {
+    let _env = IsolatedCacheEnv::new();
     unsafe {
         std::env::set_var("CRANK_CACHE_BACKEND", "valkey");
         std::env::set_var("CRANK_CACHE_URL", "redis://cache:6379/0");
@@ -34,16 +62,11 @@ fn loads_valkey_config_from_env() {
     assert_eq!(config.backend, CacheBackend::Valkey);
     assert_eq!(config.url.as_deref(), Some("redis://cache:6379/0"));
     assert_eq!(config.default_ttl_ms, Some(15_000));
-
-    unsafe {
-        std::env::remove_var("CRANK_CACHE_BACKEND");
-        std::env::remove_var("CRANK_CACHE_URL");
-        std::env::remove_var("CRANK_CACHE_DEFAULT_TTL_MS");
-    }
 }
 
 #[test]
 fn rejects_external_backend_without_url() {
+    let _env = IsolatedCacheEnv::new();
     unsafe {
         std::env::set_var("CRANK_CACHE_BACKEND", "redis");
         std::env::remove_var("CRANK_CACHE_URL");
@@ -57,14 +80,11 @@ fn rejects_external_backend_without_url() {
             backend: CacheBackend::Redis
         }
     ));
-
-    unsafe {
-        std::env::remove_var("CRANK_CACHE_BACKEND");
-    }
 }
 
 #[test]
 fn rejects_zero_ttl() {
+    let _env = IsolatedCacheEnv::new();
     unsafe {
         std::env::set_var("CRANK_CACHE_DEFAULT_TTL_MS", "0");
     }
@@ -77,9 +97,44 @@ fn rejects_zero_ttl() {
             name: "CRANK_CACHE_DEFAULT_TTL_MS"
         }
     ));
+}
 
-    unsafe {
-        std::env::remove_var("CRANK_CACHE_DEFAULT_TTL_MS");
+struct IsolatedCacheEnv {
+    _lock: MutexGuard<'static, ()>,
+    previous: Vec<(&'static str, Option<OsString>)>,
+}
+
+impl IsolatedCacheEnv {
+    fn new() -> Self {
+        let lock = CACHE_ENV_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let previous = CACHE_ENV_NAMES
+            .iter()
+            .map(|name| (*name, std::env::var_os(name)))
+            .collect();
+        for name in CACHE_ENV_NAMES {
+            unsafe {
+                std::env::remove_var(name);
+            }
+        }
+        Self {
+            _lock: lock,
+            previous,
+        }
+    }
+}
+
+impl Drop for IsolatedCacheEnv {
+    fn drop(&mut self) {
+        for (name, value) in &self.previous {
+            unsafe {
+                match value {
+                    Some(value) => std::env::set_var(name, value),
+                    None => std::env::remove_var(name),
+                }
+            }
+        }
     }
 }
 
@@ -209,6 +264,81 @@ async fn in_memory_coordination_store_scopes_keys() {
     );
 }
 
+#[tokio::test]
+async fn in_memory_coordination_store_atomically_takes_and_reserves_values() {
+    let store = InMemoryCoordinationStateStore::default();
+    let value = CoordinationStateValue {
+        payload: json!({ "state": "pending" }),
+    };
+    store
+        .put_value(
+            CacheScope::Coordination,
+            "atomic-job",
+            value.clone(),
+            Duration::from_secs(20),
+        )
+        .await
+        .unwrap();
+
+    let (first, second) = tokio::join!(
+        store.take_value(CacheScope::Coordination, "atomic-job"),
+        store.take_value(CacheScope::Coordination, "atomic-job")
+    );
+    assert_eq!(
+        usize::from(first.unwrap().is_some()) + usize::from(second.unwrap().is_some()),
+        1
+    );
+
+    assert_eq!(
+        store
+            .reserve_value(
+                CacheScope::Coordination,
+                "reservation",
+                value.clone(),
+                Duration::from_secs(20),
+            )
+            .await
+            .unwrap(),
+        CoordinationStateReservation::Reserved
+    );
+    assert_eq!(
+        store
+            .reserve_value(
+                CacheScope::Coordination,
+                "reservation",
+                CoordinationStateValue {
+                    payload: json!({ "state": "other" }),
+                },
+                Duration::from_secs(20),
+            )
+            .await
+            .unwrap(),
+        CoordinationStateReservation::Existing(value.clone())
+    );
+    let completed = CoordinationStateValue {
+        payload: json!({ "state": "completed" }),
+    };
+    assert!(
+        store
+            .compare_and_set_value(
+                CacheScope::Coordination,
+                "reservation",
+                &value,
+                completed.clone(),
+                Duration::from_secs(20),
+            )
+            .await
+            .unwrap()
+    );
+    assert_eq!(
+        store
+            .get_value(CacheScope::Coordination, "reservation")
+            .await
+            .unwrap(),
+        Some(completed)
+    );
+}
+
 #[tokio::test]
 async fn in_memory_stores_reject_empty_keys() {
     let response_store = InMemoryResponseCacheStore::default();