Split admin import and runtime safety modules

apps/admin-api/src/service.rs

@@ -41,13 +41,14 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tracing::{info, instrument};
 use uuid::Uuid;
 
+mod import_export;
+
 use crate::{
     auth::{
         AuthSettings, AuthenticatedSession, SessionCookie, create_session_cookie, hash_password,
         hash_session_secret, verify_password,
     },
     error::ApiError,
-    import_guidance::import_guidance_warnings,
     storage::LocalArtifactStorage,
 };
 
@@ -2625,129 +2626,6 @@ impl AdminService {
         Ok(())
     }
 
-    #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))]
-    pub async fn export_operation(
-        &self,
-        workspace_id: &WorkspaceId,
-        operation_id: &OperationId,
-        query: ExportQuery,
-    ) -> Result<String, ApiError> {
-        let version = match query.version {
-            Some(version) => version,
-            None => {
-                self.get_operation(workspace_id, operation_id)
-                    .await?
-                    .current_draft_version
-            }
-        };
-        let record = self
-            .get_operation_version(workspace_id, operation_id, version)
-            .await?;
-        let document = YamlOperationDocument {
-            format_version: "1".to_owned(),
-            kind: "operation".to_owned(),
-            operation: RegistryOperation {
-                config_export: Some(ConfigExport {
-                    format_version: "1".to_owned(),
-                    export_mode: query.mode,
-                }),
-                ..record.snapshot
-            },
-        };
-
-        serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string()))
-    }
-
-    #[instrument(skip(self, yaml_document), fields(mode = ?query.mode))]
-    pub async fn import_operation(
-        &self,
-        workspace_id: &WorkspaceId,
-        query: ImportQuery,
-        yaml_document: &str,
-    ) -> Result<ImportResponse, ApiError> {
-        let document: YamlOperationDocument = serde_yaml::from_str(yaml_document)
-            .map_err(|error| ApiError::validation(error.to_string()))?;
-        if document.kind != "operation" {
-            return Err(ApiError::validation("yaml kind must be operation"));
-        }
-
-        let payload = OperationPayload {
-            name: document.operation.name.clone(),
-            display_name: document.operation.display_name.clone(),
-            category: document.operation.category.clone(),
-            protocol: document.operation.protocol,
-            security_level: document.operation.security_level,
-            target: document.operation.target.clone(),
-            input_schema: document.operation.input_schema.clone(),
-            output_schema: document.operation.output_schema.clone(),
-            input_mapping: document.operation.input_mapping.clone(),
-            output_mapping: document.operation.output_mapping.clone(),
-            execution_config: document.operation.execution_config.clone(),
-            tool_description: document.operation.tool_description.clone(),
-            wizard_state: document.operation.wizard_state.clone(),
-        };
-        let warnings = import_guidance_warnings(&document.operation);
-
-        match query.mode {
-            ImportMode::Create => {
-                let created = self.create_operation(workspace_id, payload).await?;
-                Ok(ImportResponse {
-                    operation_id: created.operation_id,
-                    workspace_id: created.workspace_id,
-                    version: created.version,
-                    import_mode: ImportMode::Create,
-                    warnings,
-                })
-            }
-            ImportMode::Upsert => {
-                if let Some(existing) = self
-                    .find_operation_by_name(workspace_id, &document.operation.name)
-                    .await?
-                {
-                    let created = self
-                        .create_version(
-                            workspace_id,
-                            &existing.id,
-                            NewVersionPayload {
-                                operation: payload,
-                                change_note: Some("yaml upsert".to_owned()),
-                            },
-                        )
-                        .await?;
-
-                    let response = ImportResponse {
-                        operation_id: created.operation_id,
-                        workspace_id: created.workspace_id,
-                        version: created.version,
-                        import_mode: ImportMode::Upsert,
-                        warnings,
-                    };
-                    info!(
-                        operation_id = %response.operation_id,
-                        version = response.version,
-                        "operation imported by upsert"
-                    );
-                    Ok(response)
-                } else {
-                    let created = self.create_operation(workspace_id, payload).await?;
-                    let response = ImportResponse {
-                        operation_id: created.operation_id,
-                        workspace_id: created.workspace_id,
-                        version: created.version,
-                        import_mode: ImportMode::Upsert,
-                        warnings,
-                    };
-                    info!(
-                        operation_id = %response.operation_id,
-                        version = response.version,
-                        "operation imported by upsert"
-                    );
-                    Ok(response)
-                }
-            }
-        }
-    }
-
     #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))]
     pub async fn save_json_sample(
         &self,

apps/admin-api/src/service/import_export.rs

@@ -0,0 +1,137 @@
+use crank_core::{ConfigExport, WorkspaceId};
+use crank_registry::RegistryOperation;
+use tracing::{info, instrument};
+
+use crate::{
+    error::ApiError,
+    import_guidance::import_guidance_warnings,
+    service::{
+        AdminService, ExportQuery, ImportMode, ImportQuery, ImportResponse, NewVersionPayload,
+        OperationPayload, YamlOperationDocument,
+    },
+};
+
+impl AdminService {
+    #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))]
+    pub async fn export_operation(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &crank_core::OperationId,
+        query: ExportQuery,
+    ) -> Result<String, ApiError> {
+        let version = match query.version {
+            Some(version) => version,
+            None => {
+                self.get_operation(workspace_id, operation_id)
+                    .await?
+                    .current_draft_version
+            }
+        };
+        let record = self
+            .get_operation_version(workspace_id, operation_id, version)
+            .await?;
+        let document = YamlOperationDocument {
+            format_version: "1".to_owned(),
+            kind: "operation".to_owned(),
+            operation: RegistryOperation {
+                config_export: Some(ConfigExport {
+                    format_version: "1".to_owned(),
+                    export_mode: query.mode,
+                }),
+                ..record.snapshot
+            },
+        };
+
+        serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string()))
+    }
+
+    #[instrument(skip(self, yaml_document), fields(mode = ?query.mode))]
+    pub async fn import_operation(
+        &self,
+        workspace_id: &WorkspaceId,
+        query: ImportQuery,
+        yaml_document: &str,
+    ) -> Result<ImportResponse, ApiError> {
+        let document: YamlOperationDocument = serde_yaml::from_str(yaml_document)
+            .map_err(|error| ApiError::validation(error.to_string()))?;
+        if document.kind != "operation" {
+            return Err(ApiError::validation("yaml kind must be operation"));
+        }
+
+        let payload = OperationPayload {
+            name: document.operation.name.clone(),
+            display_name: document.operation.display_name.clone(),
+            category: document.operation.category.clone(),
+            protocol: document.operation.protocol,
+            security_level: document.operation.security_level,
+            target: document.operation.target.clone(),
+            input_schema: document.operation.input_schema.clone(),
+            output_schema: document.operation.output_schema.clone(),
+            input_mapping: document.operation.input_mapping.clone(),
+            output_mapping: document.operation.output_mapping.clone(),
+            execution_config: document.operation.execution_config.clone(),
+            tool_description: document.operation.tool_description.clone(),
+            wizard_state: document.operation.wizard_state.clone(),
+        };
+        let warnings = import_guidance_warnings(&document.operation);
+
+        match query.mode {
+            ImportMode::Create => {
+                let created = self.create_operation(workspace_id, payload).await?;
+                Ok(ImportResponse {
+                    operation_id: created.operation_id,
+                    workspace_id: created.workspace_id,
+                    version: created.version,
+                    import_mode: ImportMode::Create,
+                    warnings,
+                })
+            }
+            ImportMode::Upsert => {
+                if let Some(existing) = self
+                    .find_operation_by_name(workspace_id, &document.operation.name)
+                    .await?
+                {
+                    let created = self
+                        .create_version(
+                            workspace_id,
+                            &existing.id,
+                            NewVersionPayload {
+                                operation: payload,
+                                change_note: Some("yaml upsert".to_owned()),
+                            },
+                        )
+                        .await?;
+
+                    let response = ImportResponse {
+                        operation_id: created.operation_id,
+                        workspace_id: created.workspace_id,
+                        version: created.version,
+                        import_mode: ImportMode::Upsert,
+                        warnings,
+                    };
+                    info!(
+                        operation_id = %response.operation_id,
+                        version = response.version,
+                        "operation imported by upsert"
+                    );
+                    Ok(response)
+                } else {
+                    let created = self.create_operation(workspace_id, payload).await?;
+                    let response = ImportResponse {
+                        operation_id: created.operation_id,
+                        workspace_id: created.workspace_id,
+                        version: created.version,
+                        import_mode: ImportMode::Upsert,
+                        warnings,
+                    };
+                    info!(
+                        operation_id = %response.operation_id,
+                        version = response.version,
+                        "operation imported by upsert"
+                    );
+                    Ok(response)
+                }
+            }
+        }
+    }
+}

crates/crank-runtime/src/confirmation.rs

@@ -0,0 +1,203 @@
+use std::time::Duration;
+
+use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
+use crank_core::{
+    CacheScope, CoordinationStateStore, CoordinationStateValue, HttpMethod, OperationSafetyClass,
+    OperationSafetyPolicy, Target,
+};
+use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+use crate::{RuntimeError, RuntimeOperation, RuntimeRequestContext};
+
+pub async fn confirm_operation(
+    store: Option<&dyn CoordinationStateStore>,
+    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) = store 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, &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, &scope, provided_token, &input_hash).await
+}
+
+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()
+}

crates/crank-runtime/src/executor.rs

@@ -4,17 +4,15 @@ use std::time::{Duration, Instant};
 
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use crank_core::{
-    AdapterRegistry, CacheScope, CachedHeader, CachedResponse, CoordinationStateStore,
-    CoordinationStateValue, ExecutionMode, HttpMethod, IdempotencyMode, IdempotencyPolicy,
-    InvocationStatus, MeteringEvent, OperationSafetyClass, OperationSafetyPolicy,
-    ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter, Target,
+    AdapterRegistry, CachedHeader, CachedResponse, CoordinationStateStore, ExecutionMode,
+    HttpMethod, InvocationStatus, MeteringEvent, 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,
@@ -153,9 +151,15 @@ impl RuntimeExecutor {
         request_context: Option<&RuntimeRequestContext>,
     ) -> 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?;
+        let idempotency_key =
+            crate::idempotency::prepare_idempotency(operation, input, &mut prepared_request)?;
+        crate::confirmation::confirm_operation(
+            self.coordination_store.as_deref(),
+            operation,
+            input,
+            request_context,
+        )
+        .await?;
         if let Some(response) = self
             .load_idempotent_adapter_response(
                 operation,
@@ -202,78 +206,6 @@ impl RuntimeExecutor {
         Ok(finalized_output)
     }
 
-    fn prepare_idempotency(
-        &self,
-        operation: &RuntimeOperation,
-        input: &Value,
-        prepared_request: &mut PreparedRequest,
-    ) -> Result<Option<String>, RuntimeError> {
-        let Some(policy) = idempotency_policy(operation) else {
-            return Ok(None);
-        };
-
-        let key = idempotency_key_from_policy(policy, input, prepared_request);
-        let Some(key) = key else {
-            if policy.mode == IdempotencyMode::Required {
-                return Err(RuntimeError::InvalidPreparedRequest {
-                    field: "execution_config.idempotency".to_owned(),
-                    reason: "required idempotency key was not provided".to_owned(),
-                });
-            }
-            return Ok(None);
-        };
-
-        if let Some(header_name) = policy
-            .header_name
-            .as_deref()
-            .filter(|value| !value.is_empty())
-        {
-            prepared_request
-                .headers
-                .entry(header_name.to_owned())
-                .or_insert_with(|| key.clone());
-        }
-
-        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,
@@ -430,7 +362,8 @@ impl RuntimeExecutor {
         request_context: Option<&RuntimeRequestContext>,
     ) -> Option<AdapterResponse> {
         let response_cache = self.response_cache.as_ref()?;
-        let cache_key = idempotency_cache_key(operation, idempotency_key?, request_context)?;
+        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) => {
@@ -460,10 +393,10 @@ impl RuntimeExecutor {
         if !(200..=299).contains(&adapter_response.status_code) {
             return;
         }
-        let Some(policy) = idempotency_policy(operation) else {
+        let Some(policy) = crate::idempotency::policy(operation) else {
             return;
         };
-        let Some(cache_key) = idempotency_cache_key(
+        let Some(cache_key) = crate::idempotency::cache_key(
             operation,
             idempotency_key.unwrap_or_default(),
             request_context,
@@ -669,225 +602,6 @@ fn response_cache_ttl(operation: &RuntimeOperation) -> Option<Duration> {
     Some(Duration::from_millis(policy.ttl_ms))
 }
 
-fn idempotency_policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy> {
-    let is_mutating_rest =
-        matches!(&operation.target, Target::Rest(target) if target.method != HttpMethod::Get);
-    if !is_mutating_rest {
-        return None;
-    }
-
-    let policy = operation.execution_config.idempotency.as_ref()?;
-    if policy.mode == IdempotencyMode::Disabled || policy.ttl_ms == 0 {
-        return None;
-    }
-
-    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,
-    prepared_request: &PreparedRequest,
-) -> Option<String> {
-    if let Some(header_name) = policy
-        .header_name
-        .as_deref()
-        .filter(|value| !value.is_empty())
-    {
-        if let Some(value) = prepared_request.headers.get(header_name) {
-            return Some(value.clone());
-        }
-    }
-
-    let field_name = policy.input_field.as_deref()?.trim();
-    if field_name.is_empty() {
-        return None;
-    }
-
-    input.get(field_name).and_then(|value| match value {
-        Value::String(value) if !value.trim().is_empty() => Some(value.clone()),
-        Value::Number(value) => Some(value.to_string()),
-        _ => None,
-    })
-}
-
-fn idempotency_cache_key(
-    operation: &RuntimeOperation,
-    idempotency_key: &str,
-    request_context: Option<&RuntimeRequestContext>,
-) -> Option<String> {
-    if idempotency_key.is_empty() {
-        return None;
-    }
-
-    let scope = request_context?.response_cache_scope()?;
-    let key_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(idempotency_key.as_bytes()));
-
-    Some(format!(
-        "crank:idempotency:workspace:{}:agent:{}:operation:{}:version:{}:key:{}",
-        scope.workspace_key,
-        scope.agent_key,
-        operation.operation_id.as_str(),
-        operation.operation_version,
-        key_hash
-    ))
-}
-
 fn response_cache_key(
     operation: &RuntimeOperation,
     prepared_request: &PreparedRequest,

crates/crank-runtime/src/idempotency.rs

@@ -0,0 +1,104 @@
+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 crate::{PreparedRequest, RuntimeError, RuntimeOperation, RuntimeRequestContext};
+
+pub fn prepare_idempotency(
+    operation: &RuntimeOperation,
+    input: &Value,
+    prepared_request: &mut PreparedRequest,
+) -> Result<Option<String>, RuntimeError> {
+    let Some(policy) = policy(operation) else {
+        return Ok(None);
+    };
+
+    let key = key_from_policy(policy, input, prepared_request);
+    let Some(key) = key else {
+        if policy.mode == IdempotencyMode::Required {
+            return Err(RuntimeError::InvalidPreparedRequest {
+                field: "execution_config.idempotency".to_owned(),
+                reason: "required idempotency key was not provided".to_owned(),
+            });
+        }
+        return Ok(None);
+    };
+
+    if let Some(header_name) = policy
+        .header_name
+        .as_deref()
+        .filter(|value| !value.is_empty())
+    {
+        prepared_request
+            .headers
+            .entry(header_name.to_owned())
+            .or_insert_with(|| key.clone());
+    }
+
+    Ok(Some(key))
+}
+
+pub fn policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy> {
+    let is_mutating_rest =
+        matches!(&operation.target, Target::Rest(target) if target.method != HttpMethod::Get);
+    if !is_mutating_rest {
+        return None;
+    }
+
+    let policy = operation.execution_config.idempotency.as_ref()?;
+    if policy.mode == IdempotencyMode::Disabled || policy.ttl_ms == 0 {
+        return None;
+    }
+
+    Some(policy)
+}
+
+pub fn cache_key(
+    operation: &RuntimeOperation,
+    idempotency_key: &str,
+    request_context: Option<&RuntimeRequestContext>,
+) -> Option<String> {
+    if idempotency_key.is_empty() {
+        return None;
+    }
+
+    let scope = request_context?.response_cache_scope()?;
+    let key_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(idempotency_key.as_bytes()));
+
+    Some(format!(
+        "crank:idempotency:workspace:{}:agent:{}:operation:{}:version:{}:key:{}",
+        scope.workspace_key,
+        scope.agent_key,
+        operation.operation_id.as_str(),
+        operation.operation_version,
+        key_hash
+    ))
+}
+
+fn key_from_policy(
+    policy: &IdempotencyPolicy,
+    input: &Value,
+    prepared_request: &PreparedRequest,
+) -> Option<String> {
+    if let Some(header_name) = policy
+        .header_name
+        .as_deref()
+        .filter(|value| !value.is_empty())
+    {
+        if let Some(value) = prepared_request.headers.get(header_name) {
+            return Some(value.clone());
+        }
+    }
+
+    let field_name = policy.input_field.as_deref()?.trim();
+    if field_name.is_empty() {
+        return None;
+    }
+
+    input.get(field_name).and_then(|value| match value {
+        Value::String(value) if !value.trim().is_empty() => Some(value.clone()),
+        Value::Number(value) => Some(value.to_string()),
+        _ => None,
+    })
+}

crates/crank-runtime/src/lib.rs

@@ -1,9 +1,11 @@
 mod auth;
 mod cache;
 mod cache_factory;
+mod confirmation;
 mod error;
 mod executor;
 mod executor_builder;
+mod idempotency;
 mod limits;
 mod model;
 mod rate_limit;