Add destructive confirmation and import guidance

crates/crank-runtime/Cargo.toml

@@ -28,6 +28,7 @@ thiserror.workspace = true
 time.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tracing.workspace = true
+uuid.workspace = true
 
 [dev-dependencies]
 axum.workspace = true

crates/crank-runtime/src/error.rs

@@ -1,5 +1,5 @@
 use crank_adapter_rest::RestAdapterError;
-use crank_core::{ExecutionMode, Protocol};
+use crank_core::{ExecutionMode, OperationSafetyClass, Protocol};
 use crank_mapping::MappingError;
 use crank_schema::SchemaError;
 use thiserror::Error;
@@ -25,6 +25,19 @@ pub enum RuntimeError {
     ConcurrencyLimitExceeded { kind: &'static str, limit: usize },
     #[error("invalid prepared request at {field}: {reason}")]
     InvalidPreparedRequest { field: String, reason: String },
+    #[error(
+        "operation {operation_id} requires confirmation for {safety_class:?} action; retry with confirmation token {confirmation_token}"
+    )]
+    ConfirmationRequired {
+        operation_id: String,
+        safety_class: OperationSafetyClass,
+        confirmation_token: String,
+        expires_in_ms: u64,
+    },
+    #[error("confirmation token for operation {operation_id} is invalid, expired, or already used")]
+    InvalidConfirmationToken { operation_id: String },
+    #[error("confirmation store is unavailable for operation {operation_id}")]
+    ConfirmationStoreUnavailable { 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

@@ -4,15 +4,17 @@ use std::time::{Duration, Instant};
 
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use crank_core::{
-    AdapterRegistry, CachedHeader, CachedResponse, ExecutionMode, HttpMethod, IdempotencyMode,
-    IdempotencyPolicy, InvocationStatus, MeteringEvent, ResponseCacheStore, SharedMeteringSink,
-    SharedProtocolAdapter, Target,
+    AdapterRegistry, CacheScope, CachedHeader, CachedResponse, CoordinationStateStore,
+    CoordinationStateValue, ExecutionMode, HttpMethod, IdempotencyMode, IdempotencyPolicy,
+    InvocationStatus, MeteringEvent, OperationSafetyClass, OperationSafetyPolicy,
+    ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter, Target,
 };
 use serde_json::{Map, Value, json};
 use sha2::{Digest, Sha256};
 use time::OffsetDateTime;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tracing::debug;
+use uuid::Uuid;
 
 use crate::{
     AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeLimits, RuntimeOperation,
@@ -25,6 +27,7 @@ pub struct RuntimeExecutor {
     limits: RuntimeLimits,
     unary_limiter: Arc<Semaphore>,
     response_cache: Option<Arc<dyn ResponseCacheStore>>,
+    coordination_store: Option<Arc<dyn CoordinationStateStore>>,
     metering_sink: SharedMeteringSink,
 }
 
@@ -49,6 +52,7 @@ impl RuntimeExecutor {
         limits: RuntimeLimits,
         adapters: AdapterRegistry,
         response_cache: Option<Arc<dyn ResponseCacheStore>>,
+        coordination_store: Option<Arc<dyn CoordinationStateStore>>,
         metering_sink: SharedMeteringSink,
     ) -> Self {
         Self {
@@ -56,6 +60,7 @@ impl RuntimeExecutor {
             unary_limiter: Arc::new(Semaphore::new(limits.max_concurrent_unary)),
             limits,
             response_cache,
+            coordination_store,
             metering_sink,
         }
     }
@@ -68,6 +73,14 @@ impl RuntimeExecutor {
         self
     }
 
+    pub fn with_coordination_store(
+        mut self,
+        coordination_store: Arc<dyn CoordinationStateStore>,
+    ) -> Self {
+        self.coordination_store = Some(coordination_store);
+        self
+    }
+
     pub async fn execute(
         &self,
         operation: &RuntimeOperation,
@@ -141,6 +154,8 @@ impl RuntimeExecutor {
     ) -> Result<Value, RuntimeError> {
         let mut prepared_request = prepared_request;
         let idempotency_key = self.prepare_idempotency(operation, input, &mut prepared_request)?;
+        self.confirm_destructive_operation(operation, input, request_context)
+            .await?;
         if let Some(response) = self
             .load_idempotent_adapter_response(
                 operation,
@@ -222,6 +237,43 @@ impl RuntimeExecutor {
         Ok(Some(key))
     }
 
+    async fn confirm_destructive_operation(
+        &self,
+        operation: &RuntimeOperation,
+        input: &Value,
+        request_context: Option<&RuntimeRequestContext>,
+    ) -> Result<(), RuntimeError> {
+        let safety = effective_safety_policy(operation);
+        if !safety.class.requires_confirmation() {
+            return Ok(());
+        }
+
+        let Some(store) = self.coordination_store.as_ref() else {
+            return Err(RuntimeError::ConfirmationStoreUnavailable {
+                operation_id: operation.operation_id.as_str().to_owned(),
+            });
+        };
+
+        let scope = confirmation_scope(operation, request_context)?;
+        let input_hash = hash_json(input)?;
+        let provided_token = request_context
+            .and_then(RuntimeRequestContext::confirmation_token)
+            .or_else(|| confirmation_token_from_input(input));
+
+        let Some(provided_token) = provided_token else {
+            let token =
+                issue_confirmation_token(store.as_ref(), &scope, &input_hash, &safety).await?;
+            return Err(RuntimeError::ConfirmationRequired {
+                operation_id: operation.operation_id.as_str().to_owned(),
+                safety_class: safety.class,
+                confirmation_token: token,
+                expires_in_ms: confirmation_ttl_ms(&safety),
+            });
+        };
+
+        consume_confirmation_token(store.as_ref(), &scope, provided_token, &input_hash).await
+    }
+
     async fn record_metering<T>(
         &self,
         operation: &RuntimeOperation,
@@ -632,6 +684,161 @@ fn idempotency_policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy
     Some(policy)
 }
 
+fn effective_safety_policy(operation: &RuntimeOperation) -> OperationSafetyPolicy {
+    operation
+        .execution_config
+        .safety
+        .clone()
+        .unwrap_or_else(|| OperationSafetyPolicy {
+            class: infer_safety_class(operation),
+            confirmation: None,
+        })
+}
+
+fn infer_safety_class(operation: &RuntimeOperation) -> OperationSafetyClass {
+    match &operation.target {
+        Target::Rest(target) => match target.method {
+            HttpMethod::Get => OperationSafetyClass::Read,
+            HttpMethod::Delete => OperationSafetyClass::Destructive,
+            HttpMethod::Post | HttpMethod::Put | HttpMethod::Patch => OperationSafetyClass::Write,
+        },
+    }
+}
+
+fn confirmation_ttl_ms(policy: &OperationSafetyPolicy) -> u64 {
+    policy
+        .confirmation
+        .as_ref()
+        .map(|confirmation| confirmation.ttl_ms)
+        .filter(|ttl_ms| *ttl_ms > 0)
+        .unwrap_or(300_000)
+}
+
+fn confirmation_scope(
+    operation: &RuntimeOperation,
+    request_context: Option<&RuntimeRequestContext>,
+) -> Result<String, RuntimeError> {
+    let Some(scope) = request_context.and_then(RuntimeRequestContext::response_cache_scope) else {
+        return Err(RuntimeError::InvalidPreparedRequest {
+            field: "runtime_context.response_cache_scope".to_owned(),
+            reason: "confirmation requires workspace and agent scope".to_owned(),
+        });
+    };
+
+    Ok(format!(
+        "workspace:{}:agent:{}:operation:{}:version:{}",
+        scope.workspace_key,
+        scope.agent_key,
+        operation.operation_id.as_str(),
+        operation.operation_version
+    ))
+}
+
+async fn issue_confirmation_token(
+    store: &dyn CoordinationStateStore,
+    scope: &str,
+    input_hash: &str,
+    safety: &OperationSafetyPolicy,
+) -> Result<String, RuntimeError> {
+    let token = format!("ct_{}", Uuid::now_v7().simple());
+    let key = confirmation_cache_key(scope, &token);
+    let ttl_ms = confirmation_ttl_ms(safety);
+    store
+        .put_value(
+            CacheScope::Coordination,
+            &key,
+            CoordinationStateValue {
+                payload: json!({
+                    "scope": scope,
+                    "input_hash": input_hash,
+                }),
+            },
+            Duration::from_millis(ttl_ms),
+        )
+        .await
+        .map_err(|error| RuntimeError::InvalidPreparedRequest {
+            field: "confirmation_token".to_owned(),
+            reason: error.to_string(),
+        })?;
+    Ok(token)
+}
+
+async fn consume_confirmation_token(
+    store: &dyn CoordinationStateStore,
+    operation_scope: &str,
+    token: &str,
+    input_hash: &str,
+) -> Result<(), RuntimeError> {
+    let key = confirmation_cache_key(operation_scope, token);
+    let stored = store
+        .get_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 {
+            operation_id: operation_id_from_scope(operation_scope),
+        });
+    };
+
+    let matches_scope =
+        stored.payload.get("scope").and_then(Value::as_str) == Some(operation_scope);
+    let matches_input =
+        stored.payload.get("input_hash").and_then(Value::as_str) == Some(input_hash);
+
+    if matches_scope && matches_input {
+        Ok(())
+    } else {
+        Err(RuntimeError::InvalidConfirmationToken {
+            operation_id: operation_id_from_scope(operation_scope),
+        })
+    }
+}
+
+fn confirmation_cache_key(scope: &str, token: &str) -> String {
+    let token_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(token.as_bytes()));
+    format!("crank:confirmation:{scope}:token:{token_hash}")
+}
+
+fn confirmation_token_from_input(input: &Value) -> Option<&str> {
+    input
+        .get("_crank_confirmation_token")
+        .and_then(Value::as_str)
+        .filter(|value| !value.trim().is_empty())
+}
+
+fn hash_json(input: &Value) -> Result<String, RuntimeError> {
+    let sanitized = input_without_confirmation_token(input);
+    let encoded =
+        serde_json::to_vec(&sanitized).map_err(|error| RuntimeError::InvalidPreparedRequest {
+            field: "confirmation_input".to_owned(),
+            reason: error.to_string(),
+        })?;
+    Ok(URL_SAFE_NO_PAD.encode(Sha256::digest(encoded)))
+}
+
+fn input_without_confirmation_token(input: &Value) -> Value {
+    let Value::Object(object) = input else {
+        return input.clone();
+    };
+    let mut sanitized = object.clone();
+    sanitized.remove("_crank_confirmation_token");
+    Value::Object(sanitized)
+}
+
+fn operation_id_from_scope(scope: &str) -> String {
+    scope
+        .split(":operation:")
+        .nth(1)
+        .and_then(|tail| tail.split(":version:").next())
+        .unwrap_or("<unknown>")
+        .to_owned()
+}
+
 fn idempotency_key_from_policy(
     policy: &IdempotencyPolicy,
     input: &Value,

crates/crank-runtime/src/executor_builder.rs

@@ -2,8 +2,8 @@ use std::sync::Arc;
 
 use crank_adapter_rest::RestAdapter;
 use crank_core::{
-    AdapterRegistry, MeteringSink, NoopMeteringSink, ResponseCacheStore, SharedMeteringSink,
-    SharedProtocolAdapter,
+    AdapterRegistry, CoordinationStateStore, MeteringSink, NoopMeteringSink, ResponseCacheStore,
+    SharedMeteringSink, SharedProtocolAdapter,
 };
 
 use crate::{RuntimeExecutor, RuntimeLimits};
@@ -12,6 +12,7 @@ pub struct RuntimeExecutorBuilder {
     limits: RuntimeLimits,
     adapters: AdapterRegistry,
     response_cache: Option<Arc<dyn ResponseCacheStore>>,
+    coordination_store: Option<Arc<dyn CoordinationStateStore>>,
     metering_sink: SharedMeteringSink,
 }
 
@@ -21,6 +22,7 @@ impl RuntimeExecutorBuilder {
             limits: RuntimeLimits::default(),
             adapters: AdapterRegistry::empty(),
             response_cache: None,
+            coordination_store: None,
             metering_sink: Arc::new(NoopMeteringSink) as Arc<dyn MeteringSink>,
         }
     }
@@ -40,6 +42,11 @@ impl RuntimeExecutorBuilder {
         self
     }
 
+    pub fn with_coordination_store(mut self, store: Arc<dyn CoordinationStateStore>) -> Self {
+        self.coordination_store = Some(store);
+        self
+    }
+
     pub fn with_metering_sink(mut self, sink: SharedMeteringSink) -> Self {
         self.metering_sink = sink;
         self
@@ -50,6 +57,7 @@ impl RuntimeExecutorBuilder {
             self.limits,
             self.adapters,
             self.response_cache,
+            self.coordination_store,
             self.metering_sink,
         )
     }

crates/crank-runtime/src/request_context.rs

@@ -14,6 +14,7 @@ pub struct RuntimeRequestContext {
     pub correlation_id: String,
     pub response_cache_scope: Option<ResponseCacheScope>,
     pub metering_context: Option<MeteringContext>,
+    pub confirmation_token: Option<String>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -30,6 +31,7 @@ impl RuntimeRequestContext {
             correlation_id: correlation_id.into(),
             response_cache_scope: None,
             metering_context: None,
+            confirmation_token: None,
         }
     }
 
@@ -78,6 +80,15 @@ impl RuntimeRequestContext {
     pub fn metering_context(&self) -> Option<&MeteringContext> {
         self.metering_context.as_ref()
     }
+
+    pub fn with_confirmation_token(mut self, token: impl Into<String>) -> Self {
+        self.confirmation_token = Some(token.into());
+        self
+    }
+
+    pub fn confirmation_token(&self) -> Option<&str> {
+        self.confirmation_token.as_deref()
+    }
 }
 
 impl From<ResponseCacheScope> for crank_core::ResponseCacheScope {
@@ -142,6 +153,7 @@ mod tests {
         assert_eq!(context.request_id, "req_123");
         assert_eq!(context.correlation_id, "req_123");
         assert!(context.response_cache_scope.is_none());
+        assert!(context.confirmation_token.is_none());
     }
 
     #[test]
@@ -167,4 +179,12 @@ mod tests {
         assert_eq!(metering.agent_id, None);
         assert_eq!(metering.source, InvocationSource::AdminTestRun);
     }
+
+    #[test]
+    fn attaches_confirmation_token() {
+        let context =
+            RuntimeRequestContext::from_request_id("req_123").with_confirmation_token("ct_123");
+
+        assert_eq!(context.confirmation_token(), Some("ct_123"));
+    }
 }

crates/crank-runtime/tests/integration.rs

@@ -1,4 +1,5 @@
 mod integration {
+    mod confirmation;
     mod idempotency;
     mod no_input_get;
 }

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

@@ -0,0 +1,222 @@
+use std::collections::BTreeMap;
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
+
+use async_trait::async_trait;
+use crank_core::{
+    AdapterResponse, ConfirmationPolicy, ExecutionConfig, ExecutionMode, HttpMethod, 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 serde_json::json;
+use time::OffsetDateTime;
+
+#[tokio::test]
+async fn destructive_operation_requires_one_time_confirmation_token() {
+    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 = destructive_delete_operation().into();
+    let context = RuntimeRequestContext::from_request_id("req_confirm")
+        .with_response_cache_scope("workspace_1", "agent_1");
+
+    let first = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "order_id": "ord_123" }),
+            Some(&context),
+        )
+        .await
+        .expect_err("first destructive call must only stage confirmation");
+    let confirmation_token = match first {
+        RuntimeError::ConfirmationRequired {
+            confirmation_token, ..
+        } => confirmation_token,
+        error => panic!("expected confirmation required, got {error:?}"),
+    };
+    assert_eq!(call_count.load(Ordering::SeqCst), 0);
+
+    let confirmed_context = context
+        .clone()
+        .with_confirmation_token(confirmation_token.clone());
+    let confirmed = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "order_id": "ord_123" }),
+            Some(&confirmed_context),
+        )
+        .await
+        .unwrap();
+    assert_eq!(confirmed, json!({ "deleted": true }));
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+
+    let replay = executor
+        .execute_with_context(
+            &operation,
+            &json!({ "order_id": "ord_123" }),
+            Some(&confirmed_context),
+        )
+        .await
+        .expect_err("confirmation token must be one-time");
+    assert!(matches!(
+        replay,
+        RuntimeError::InvalidConfirmationToken { .. }
+    ));
+    assert_eq!(call_count.load(Ordering::SeqCst), 1);
+}
+
+struct CountingAdapter {
+    call_count: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl ProtocolAdapter for CountingAdapter {
+    fn protocol(&self) -> Protocol {
+        Protocol::Rest
+    }
+
+    fn supports_mode(&self, mode: ExecutionMode) -> bool {
+        mode == ExecutionMode::Unary
+    }
+
+    async fn invoke_unary(
+        &self,
+        _target: &Target,
+        prepared: &crank_core::PreparedRequest,
+        _context: &crank_core::RuntimeRequestContext,
+    ) -> Result<AdapterResponse, ProtocolAdapterError> {
+        assert_eq!(
+            prepared.path_params.get("order_id"),
+            Some(&"ord_123".to_owned())
+        );
+        self.call_count.fetch_add(1, Ordering::SeqCst);
+        Ok(AdapterResponse {
+            status_code: 200,
+            headers: BTreeMap::new(),
+            body: json!({ "deleted": true }),
+            data: json!({ "deleted": true }),
+        })
+    }
+}
+
+fn destructive_delete_operation() -> Operation<Schema, MappingSet> {
+    Operation {
+        id: OperationId::new("op_delete_order"),
+        name: "delete_order".to_owned(),
+        display_name: "Delete order".to_owned(),
+        category: "orders".to_owned(),
+        protocol: Protocol::Rest,
+        security_level: OperationSecurityLevel::Standard,
+        status: OperationStatus::Published,
+        version: 1,
+        target: Target::Rest(RestTarget {
+            base_url: "https://api.example.invalid".to_owned(),
+            method: HttpMethod::Delete,
+            path_template: "/orders/{order_id}".to_owned(),
+            static_headers: BTreeMap::new(),
+        }),
+        input_schema: object_schema(BTreeMap::from([("order_id".to_owned(), string_schema())])),
+        output_schema: object_schema(BTreeMap::from([("deleted".to_owned(), bool_schema())])),
+        input_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.mcp.order_id".to_owned(),
+                target: "$.request.path.order_id".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        output_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.response.body.deleted".to_owned(),
+                target: "$.output.deleted".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: Some(OperationSafetyPolicy {
+                class: OperationSafetyClass::Destructive,
+                confirmation: Some(ConfirmationPolicy { ttl_ms: 60_000 }),
+            }),
+            auth_profile_ref: None,
+            headers: BTreeMap::new(),
+        },
+        tool_description: ToolDescription {
+            title: "Delete order".to_owned(),
+            description: "Deletes an order after explicit confirmation.".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,
+        description: None,
+        required: true,
+        nullable: false,
+        default_value: None,
+        fields: BTreeMap::new(),
+        items: None,
+        enum_values: Vec::new(),
+        variants: Vec::new(),
+    }
+}

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

@@ -145,6 +145,7 @@ fn idempotent_post_operation() -> Operation<Schema, MappingSet> {
                 input_field: Some("request_id".to_owned()),
                 header_name: Some("Idempotency-Key".to_owned()),
             }),
+            safety: None,
             auth_profile_ref: None,
             headers: BTreeMap::new(),
         },

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

@@ -70,6 +70,7 @@ fn no_input_get_operation() -> Operation<Schema, MappingSet> {
             retry_policy: None,
             response_cache: None,
             idempotency: None,
+            safety: None,
             auth_profile_ref: None,
             headers: BTreeMap::new(),
         },